Amazon Bedrock AgentCore로 AWS Abuse 사고 조사를 자동화하기
들어가며
MSP 환경에서 여러 Payer 계정을 운영하다 보면 AWS Health를 통해 Abuse/Security 이벤트가 수시로 날아옵니다. 자격증명 유출 의심, 포트 스캐닝, SES 남용, 콘솔 침해 의심… 이벤트 자체는 짧은 메시지 한 줄이지만, 실제로 대응하려면 담당자가 해당 계정에 들어가서 CloudTrail을 뒤지고, 어떤 리소스가 만들어졌는지 리전별로 확인하고, 타임라인을 정리해서 보고서를 써야 합니다. 한 건 조사하는 데 짧게는 몇십 분, 길게는 반나절이 걸리는 작업입니다.
“이 조사 과정을 AI 에이전트한테 시키면 어떨까?” 라는 생각에서 시작해서, Amazon Bedrock AgentCore 위에 자율 조사 에이전트를 올리고, Health 이벤트 수집부터 Slack 알림, 조사 보고서 생성까지 이어지는 어뷰즈 대응 파이프라인을 만들었습니다. 이번 글에서는 그중에서도 Bedrock AgentCore를 어떻게 활용했는지를 중심으로 정리해 보겠습니다.
왜 Bedrock AgentCore였나
처음에는 Lambda에 에이전트를 올리는 것도 고려했습니다. 하지만 침해 조사라는 작업의 특성상 몇 가지 벽에 부딪혔습니다.
- 실행 시간: CloudTrail을 여러 리전에서 뒤지고, LLM이 단서를 따라가며 추가 조회를 반복하면 조사 한 건에 20분 이상 걸립니다. Lambda의 15분 제한으로는 부족합니다.
- 도구 실행 환경: 에이전트가 AWS API를 자유롭게 조회하려면 MCP 서버 같은 서브프로세스를 띄울 수 있는 컨테이너 환경이 필요했습니다.
- 세션 격리와 상태: 조사 실행마다 격리된 세션이 필요하고, “이 계정, 예전에도 털린 적 있지 않았나?” 같은 과거 기억도 필요했습니다.
AgentCore는 이 요구사항에 정확히 맞아떨어졌습니다.
| 요구사항 | AgentCore 기능 |
|---|---|
| 장시간 실행 (20분+) | Runtime — 컨테이너 기반, 최대 8시간 실행, 세션 타임아웃 설정 가능 |
| AWS API 자유 조회 | 컨테이너 안에서 MCP 서버를 서브프로세스로 직접 실행 |
| 재발 탐지 (과거 기억) | Memory — LTM + Semantic 전략 |
| 트레이싱 | 기본 제공 Observability (OpenTelemetry) |
프레임워크는 AWS의 오픈소스 에이전트 SDK인 Strands Agents를 사용했고, 모델은 Claude Sonnet 4.5를 사용했습니다. AgentCore는 프레임워크에 중립적이라 Strands든 LangGraph든 컨테이너에 담기만 하면 됩니다.
전체 아키텍처
민감한 식별자는 제외하고 흐름만 그리면 아래와 같습니다.
수집/알림 쪽은 평범한 서버리스 구성이라 이 글에서는 생략하고 AgentCore Runtime 내부를 집중적으로 살펴보겠습니다.
AgentCore Runtime — 조사 에이전트의 실행 기반
1. 엔트리포인트와 비동기 실행 패턴
AgentCore Runtime은 BedrockAgentCoreApp으로 감싼 파이썬 앱을 ARM64 컨테이너로 배포하는 구조입니다. 엔트리포인트는 놀랄 만큼 단순합니다.
from bedrock_agentcore.runtime import BedrockAgentCoreApp
app = BedrockAgentCoreApp()
@app.entrypoint
def invoke(payload: dict) -> dict:
"""InvokeAgentRuntime 호출을 받는 진입점.
조사는 20분 이상 걸리므로 여기서 다 하지 않는다.
- 입력 파싱 후 백그라운드 스레드에서 조사 시작
- 즉시 ACK 응답 반환 (호출자는 기다리지 않음)
- 진행 상태는 DynamoDB, 완료 신호는 S3 마커로 전달
"""
params = json.loads(payload.get("prompt", "{}"))
thread = threading.Thread(
target=run_investigation, # 실제 조사 파이프라인
kwargs=params,
daemon=False,
)
thread.start()
return {"status": "ACCEPTED", "account_id": params["account_id"]}
여기서 중요한 설계 포인트가 하나 있습니다. InvokeAgentRuntime을 동기로 붙잡고 있으면 호출한 Lambda 쪽 타임아웃과 씨름해야 합니다. 그래서 엔트리포인트는 수 초 안에 즉시 응답하고, 실제 조사는 백그라운드 스레드에서 계속 실행되도록 했습니다. 조사가 끝나면 S3에 .status.json 마커 파일을 쓰고, Step Functions 쪽 Worker Lambda가 이 마커를 폴링해서 완료를 감지합니다.
def _write_status_marker(report_bucket: str, account_id: str, marker: dict) -> None:
"""AgentCore 완료 신호를 S3에 작성. Worker 폴링이 읽는 단일 진실(single source of truth)."""
ts = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S")
marker_key = f"reports/{account_id}/{ts}.status.json"
boto3.client("s3").put_object(
Bucket=report_bucket, Key=marker_key,
Body=json.dumps(marker, ensure_ascii=False).encode("utf-8"),
)
AgentCore Runtime의 세션이 조사 도중 끊기지 않도록, 배포 시 세션 유휴/최대 수명을 40분으로 늘려 두었습니다. 이 설정 덕분에 백그라운드 스레드가 조사를 끝까지 마칠 수 있습니다.
2. 조사 파이프라인 — Investigator → Triage → Graph → Reporter
컨테이너 안에서 돌아가는 조사 파이프라인은 4단계입니다. 각 단계의 역할을 명확히 분리한 것이 핵심입니다.
INPUT: { account_id, 조사 기간, 인시던트 메타데이터 }
│
├─ [1] Investigator ── 자율 조사관. CloudTrail에서 단서를 따라가며 증거 수집
├─ [2] Triage ── 증거마다 verdict 판정
├─ [3] Graph 실행 ── 침해 정황이 있을 때만: 리전 그룹 병렬 리소스 조사
├─ [4] Reporter ── 최종 보고서 생성 (구조화 출력)
│
└─ 검증·정화 → HTML 렌더 → S3 업로드 → 상태 마커
Investigator는 이 시스템의 심장입니다. “조사 기간 내 ConsoleLogin을 다 조회해라” 같은 고정 시나리오가 아니라, 1차 신호(낯선 국가에서의 접근, AccessDenied 반복, EnableRegion 등)를 보고 다음에 어떤 도구를 호출할지 스스로 결정하는 자율 에이전트입니다. Strands의 @tool 데코레이터로 CloudTrail 조회 도구들을 만들어 붙였습니다.
INVESTIGATOR_SYSTEM_PROMPT = """
당신은 AWS 침해 사고 조사관(Investigator)입니다.
주어진 조사 기간은 *seed analysis window*이며 확장 가능한 출발점입니다 —
단서가 발견되면 도구의 time_window 인자로 derived window를 만들어
seed 경계 밖까지 추적하세요.
...
"""
# 무한 조사 방지: 종료 조건은 LLM의 판단이 아니라 외부 budget이 강제한다
@dataclass(frozen=True)
class InvestigatorRuntime:
max_tool_calls: int = 30
max_wall_seconds: int = 300
여기서 배운 것 하나! 에이전트의 종료 조건을 LLM에게 맡기면 안 됩니다. “충분히 조사했으면 멈춰”라는 프롬프트는 어떤 날은 2번 만에 멈추고 어떤 날은 끝없이 돕니다. 도구 호출 횟수와 시간 예산을 코드에서 강제하고, 예산 소진 신호를 받으면 에이전트가 정리(wrap-up)하도록 하는 편이 훨씬 안정적이었습니다.
Triage는 Investigator가 모은 증거를 하나씩 malicious / uncertain / benign으로 판정하고 confidence(0~100)를 매깁니다. 조사(수집)와 판정(해석)을 다른 에이전트로 분리하니 “증거는 있는데 해석이 널뛰는” 문제가 크게 줄었습니다.
Graph는 Strands의 GraphBuilder로 만든 멀티 에이전트 그래프입니다. Triage에서 침해 정황이 확인된 경우에만 실행되며, 활성 리전을 그룹으로 나눠 각 에이전트가 병렬로 리소스 변경 이력을 조사합니다.
from strands.multiagent import GraphBuilder
def build_investigation_graph(region_agents, reporter):
builder = GraphBuilder()
builder.add_node(PassthroughNode("start")) # LLM 호출 없는 시작 노드
for i, agent in enumerate(region_agents, start=1): # 활성 리전 수에 따라 1~3개
builder.add_node(agent, f"region_group_{i}")
builder.add_edge("start", f"region_group_{i}") # 병렬 시작
builder.add_edge(f"region_group_{i}", "reporter") # 전원 완료 후 Reporter
builder.add_node(reporter, "reporter")
return builder.build()
┌── region_group_1 ──┐
start ────────├── region_group_2 ──┼──► reporter
└── region_group_3 ──┘
리전 그룹 에이전트들은 컨테이너 안에서 awslabs.aws-api-mcp-server를 서브 프로세스로 직접 실행해서 AWS API를 조회합니다. AgentCore가 일반 컨테이너 환경이기 때문에 가능한 구성입니다.
def create_aws_api_mcp_client(credentials: dict, region: str) -> MCPClient:
"""AWS API MCP 서버를 stdio 서브프로세스로 띄우는 클라이언트.
조사 대상 계정의 STS 임시 자격 증명을 env로 주입해서
에이전트의 조회 권한을 해당 계정·세션으로 한정한다.
"""
return MCPClient(lambda: stdio_client(
StdioServerParameters(
command="uvx",
args=["awslabs.aws-api-mcp-server@latest"],
env={
"AWS_REGION": region,
"AWS_ACCESS_KEY_ID": credentials["AccessKeyId"],
"AWS_SECRET_ACCESS_KEY": credentials["SecretAccessKey"],
"AWS_SESSION_TOKEN": credentials["SessionToken"],
"AWS_MAX_ATTEMPTS": "2", # 스로틀 시 재시도 제한 (도구 호출 지연 방지)
},
)
))
3. 환각과의 싸움 — 결정적 검증 레이어
보안 보고서에서 환각은 치명적입니다. 존재하지 않는 event ID, 엉뚱한 계정 ID, KST/UTC가 뒤섞인 시각이 보고서에 들어가면 그 보고서는 신뢰를 잃습니다. 그래서 LLM 출력과 최종 보고서 사이에 결정적(deterministic) 검증 레이어를 두었습니다.
핵심 아이디어는 AuthorityCapture 조사 중 도구가 실제로 조회한 event ID와 계정 ID를 “권위 있는 값”으로 축적해 두고, Reporter가 생성한 보고서에서 그 집합에 없는 값이 나오면 사후에 제거하는 방식입니다.
# 검증·정화 파이프라인 (LLM 출력 → 최종 보고서 사이)
report = validate_against_authority(report, authority) # 조회된 적 없는 event_id/계정 제거
report = sanitize_narrative_kst(report) # 서술형 필드 시각 표기를 UTC로 통일
report.unified_timeline = synthesize_unified_timeline( # 타임라인은 LLM이 아니라
report.trail_events, resource_changes) # 시스템이 결정적으로 합성
같은 맥락에서 공격 타임라인과 조사 기간 헤더는 아예 LLM이 생성하지 않습니다. Reporter 프롬프트에 “unified_timeline은 반드시 빈 리스트로 두라”고 명시하고, 실제 타임라인은 검증된 이벤트들을 시스템이 시간순으로 병합해 만듭니다. “LLM이 잘 쓰도록 유도”하는 것보다 “LLM이 쓸 필요가 없게 만드는” 쪽이 훨씬 확실했습니다.
4. AgentCore Memory — “이 계정, 전에도 털렸었는데?”
어뷰즈 대응에서 재발 여부는 대응 강도를 결정하는 중요한 신호입니다. 같은 계정이 한 달 안에 또 침해되었다면, 그건 자격 증명 관리에 구조적인 문제가 있다는 뜻이니까요.
이 “과거 기억”을 **AgentCore Memory (LTM + Semantic 전략)**로 구현했습니다. 조사가 끝날 때마다 침해 지문(fingerprint) 공격자 IAM 주체, 소스 IP, 국가, user agent, verdict 분포 등을 Memory에 적재하고, 다음 조사 시작 시 같은 계정의 최근 30일 지문을 꺼내 비교합니다.
# 조사 시작: 과거 30일 fingerprint 로드
past = load_recent_fingerprints(account_id, days=30)
# ... 조사 파이프라인 실행 ...
# 재발 판정: actor/IP/국가/UA 매칭 룰을 우선순위대로 적용 (결정적 판정)
recurrence = detect_recurrence(current_fp, past)
# 조사 종료: 이번 결과를 fingerprint로 적재 (다음 조사의 과거가 된다)
save_fingerprint(current_fp)
설계에서 신경 쓴 부분:
- 조사 1회 = 1 세션 ({account_id}-{timestamp}), namespace는 계정 ID 기준으로 분리
- Memory API 호출은 전부 safe wrapper로 감싸서, Memory 장애가 조사 본체를 절대 중단시키지 않도록 graceful degrade
- 재발 “판정” 자체는 LLM이 아닌 결정적 룰로 수행 Memory는 기억을 제공하고, 판단은 코드가 합니다
재발이 감지되면 보고서에 재발 배지와 과거 사례 탭이 붙고, Reporter는 결론 서두에 한 문장으로만 언급하도록 프롬프트에서 제약했습니다.
5. 배포 — starter toolkit으로 컨테이너부터 IAM까지
배포는 bedrock-agentcore-starter-toolkit을 사용했습니다. Dockerfile 빌드, ECR 푸시, Runtime 생성/업데이트, execution role 생성까지 한 번에 처리해 줍니다.
from bedrock_agentcore_starter_toolkit import Runtime
runtime = Runtime()
runtime.configure(
entrypoint="main.py",
auto_create_execution_role=True, # execution role 자동 생성
auto_create_ecr=True, # ECR 리포지토리 자동 생성
requirements_file="requirements.txt",
region="ap-northeast-2",
agent_name="strands_security_analysis",
vpc_enabled=True, # VPC 내부 실행 (내부 리소스 접근용)
idle_timeout=2400, # 세션 유휴 40분 — 장시간 조사 대응
max_lifetime=2400,
)
# 배포 실행 — 컨테이너 환경변수로 Memory 설정 주입
launch_result = runtime.launch(
auto_update_on_conflict=True,
env_vars={"ENABLE_MEMORY_RECURRENCE": "true", ...},
)
한 가지 팁! toolkit이 자동 생성하는 execution role에는 기본 권한만 있어서, 배포 스크립트 마지막에 Runtime이 필요로 하는 추가 권한(진행 상태 기록용 DynamoDB, Memory API 등)을 인라인 정책으로 부착하는 단계를 넣었습니다. 또 배포 전에 코드가 의존하는 테이블이 실제로 존재하는지 describe_table로 fail-fast 검증합니다. “인프라 배포가 먼저”라는 순서를 사람의 기억이 아니라 배포 스크립트가 강제하도록요.
# 예: Runtime이 AgentCore Memory를 쓰기 위한 최소 권한 (일부)
{
"Effect": "Allow",
"Action": [
"bedrock-agentcore:CreateEvent", # fingerprint 적재
"bedrock-agentcore:RetrieveMemoryRecords", # 과거 fingerprint 조회
],
"Resource": "*",
}
6. 관측 — 에이전트를 어떻게 들여다볼까
에이전트는 “지금 뭘 하고 있는지”가 보이지 않으면 운영이 불가능합니다. 세 겹으로 관측을 구성했습니다.
- AgentCore Observability — 컨테이너에 aws-opentelemetry-distro를 넣으면 LLM 호출, 도구 호출 트레이스가 CloudWatch에 자동으로 쌓입니다. “Investigator가 어떤 도구를 어떤 순서로 불렀는지”를 콘솔에서 그대로 볼 수 있습니다.
- 진행 단계 상태 — 파이프라인 단계마다 DynamoDB의 job 레코드를 갱신(CLOUDTRAIL_SCANNING → AGGREGATING → RESOURCE_SCANNING → REPORTING)하고, 대시보드 프로그레스 바가 이 값을 폴링합니다. 사용자가 “지금 몇 단계인지”를 실시간으로 봅니다.
- Step Functions 실행 그래프 — 호출 → 폴링 → 완료 처리의 외부 흐름은 Step Functions 콘솔에서 실행 단위로 추적합니다.
완성된 어뷰즈 대응 절차
이 모든 조각을 이으면, 어뷰즈 대응 절차는 이렇게 바뀝니다.
Before
Health 이벤트 확인 → 담당자가 계정 접속 → CloudTrail 수동 조회 → 리전별 리소스 확인 → 엑셀/문서로 타임라인 정리 → 보고서 작성 (수 시간)
After
Slack 알림 확인 → 대시보드에서 인시던트 클릭 → [보고서 생성] 버튼 → HTML 보고서 수령
보고서에는 사고 개요(verdict 분포 포함), 리전 옵트인/아웃 변화, 침해 의심 이벤트, 공격 타임라인, 피해 리소스 상세, 조치 필요 사항, 결론이 탭으로 정리되고, 재발 이력이 있으면 재발 배지가 붙습니다. 조사 결과가 “침해 정황 없음”이면 LLM 보고서 대신 결정적인 NO_THREAT 보고서가 나가서 비용도 아낍니다.
마무리 — AgentCore를 써보고 느낀 것
이번에 얻은 교훈을 정리하면:
- AgentCore Runtime은 “장시간 + 도구 실행”이 필요한 에이전트에 적합합니다. Lambda의 15분 벽, 서브프로세스 실행 제약에서 벗어나면서도 서버 관리는 없습니다.
- 에이전트에게 자율성을 주되, 울타리는 코드로 칩니다. 조사 경로는 LLM이 정하지만, 종료 조건(budget)·출력 검증(authority)·타임라인 합성은 전부 결정적 코드가 담당합니다. “LLM이 잘하게 만들기”보다 “LLM이 틀릴 수 없게 만들기”가 보안 도메인에서는 정답에 가까웠습니다.
- Memory는 에이전트를 ‘조사 도구’에서 ‘조사 이력을 가진 동료’로 바꿔 줍니다. 재발 탐지 하나만으로도 보고서의 실무 가치가 눈에 띄게 올라갔습니다.
침해 조사처럼 “정해진 절차 + 상황 판단”이 섞인 운영 업무는 에이전트 자동화와 궁합이 좋습니다. 비슷한 고민을 하고 계시다면 Bedrock AgentCore로 작게 시작해 보시길 권합니다.
참고 자료
Amazon Bedrock AgentCore 공식 문서
bedrock-agentcore-starter-toolkit (PyPI)
글 | 메가존클라우드 Commercial Managed Unit 최영훈 매니저
![[고객사례] 경농 | 생성형 AI로 스마트 영농일지 ‘파밍노트’의 혁신을 끌어올리다](https://img.iting.co.kr/wp-content/uploads/2026/08/%EA%B2%BD%EB%86%8D-mzc-logo-scaled.png){kind=logo}
