from agent_memory_guard import MemoryGuard
from langchain.memory import ConversationBufferMemory

guard = MemoryGuard()

class GuardedMemory(ConversationBufferMemory):
    def save_context(self, inputs, outputs):
        for content in [*inputs.values(), *outputs.values()]:
            result = guard.scan(str(content))
            if not result.is_safe:
                raise SecurityError(f"Memory poisoning blocked: {result.threat_type}")
        super().save_context(inputs, outputs)

# 즉시 교체 가능한 메모리
memory = GuardedMemory()
agent = initialize_agent(tools, llm, memory=memory)

from agent_memory_guard import MemoryGuard
from langgraph.checkpoint.memory import MemorySaver

guard = MemoryGuard()

class GuardedCheckpointer(MemorySaver):
    async def aput(self, config, checkpoint, metadata, new_versions):
        for key, value in checkpoint.get("channel_values", {}).items():
            result = guard.scan(str(value))
            if not result.is_safe:
                raise SecurityError(f"Blocked in '{key}': {result.threat_type}")
        return await super().aput(config, checkpoint, metadata, new_versions)

# 그래프에서 사용
graph = builder.compile(checkpointer=GuardedCheckpointer())

from agent_memory_guard import MemoryGuard
from autogen import ConversableAgent

guard = MemoryGuard()

class GuardedAgent(ConversableAgent):
    def _process_received_message(self, message, sender, silent):
        if isinstance(message, dict):
            content = message.get("content", "")
        else:
            content = str(message)
            
        result = guard.scan(content)
        if not result.is_safe:
            # 예외를 발생시키는 대신 로그를 남기고 격리함
            print(f"Memory poisoning attempt blocked: {result.threat_type}")
            return
        # 오염된 메시지를 저장하지 않음
        super().

def safe_add(content: str, user_id: str): result = guard.scan(content) if result.is_safe: mem0.add(content, user_id=user_id) else: raise SecurityError(f"Blocked: {result.threat_type}")
def safe_search(query: str, user_id: str): memories = mem0.search(query, user_id=user_id)
# 반환하기 전에 검색된 메모리 유효성 검사 return [m for m in memories if guard.scan(m['memory']).is_safe]

Any Framework (Generic Pattern) 만약 사용 중인 프레임워크가 위에 나열되지 않았다면, 패턴은 항상 동일합니다:
from agent_memory_guard import MemoryGuard
guard = MemoryGuard()
# 1. 쓰기 작업 래핑 def safe_memory_write(content: str):
result = guard.scan(content)
if not result.is_safe:
    raise SecurityError(f"Blocked: {result.threat_type}")
your_framework.memory.write(content)
# 2. 선택적으로 읽기 시 유효성 검사 def safe_memory_read(query: str):
memories = your_framework.memory.read(query)
return [m for m in memories if guard.scan(str(m)).is_safe]

Advanced: Guard 구성하기 기본 설정은 엄격합니다. 프로덕션 환경에서는 설정을 조정하고 싶을 수 있습니다:
from agent_memory_guard import MemoryGuard, GuardConfig
config = GuardConfig(
    # 민감도: 0.0 (허용적)부터 1.0 (엄격함)까지 sensitivity = 0.7,
    # 위반 시 동작: "raise", "quarantine", 또는 "log_only" on_violation = "quarantine",
    # 특정 탐지기 활성화/비활성화 enable_semantic_similarity = True, enable_pattern_matching = True ,
    # 감사 로깅 audit_log_path = "/var/log/agent_memory_guard.jsonl"
)
guard = MemoryGuard(config=config)

Why This Matters Now 지금 이것이 중요한 이유 OWASP Top 10 for Agentic AI Systems는 메모리 오염(memory poisoning)을 ASI06으로 등재했으며, 이는 이론적인 문제가 아닙니다.

에이전트가 데모 단계에서 프로덕션(production) 단계로 넘어가면서 다음과 같은 변화가 나타나고 있습니다:

- 더 많은 에이전트가 영구 메모리(persistent memory)를 보유함 (RAG, 벡터 저장소(vector stores), 대화 기록(conversation history))
- 더 많은 에이전트가 여러 세션에 걸쳐 자율적으로 작동함
- 더 많은 에이전트가 민감한 작업(APides, 데이터베이스(databases), 파일 시스템(file systems))에 접근함

공격 표면(attack surface)이 방어 체계보다 더 빠르게 확장되고 있습니다. 메모리 오염(memory poisoning)은 다음과 같은 특징을 가진 몇 안 되는 공격 중 하나입니다:

- 공격자의 지속적인 접근 권한이 필요하지 않음
- 보안 업데이트 및 재시작 후에도 지속됨
- 표준 모니터링(standard monitoring)으로는 탐지할 수 없음

시작하기
pip install agent-memory-guard
OWASP 프로젝트: github.com/OWASP/www-project-agent-memory-guard

만약 영구 메모리를 사용하는 프로덕션용 AI 에이전트를 구축하고 계신다면, 이러한 공격 표면에 대해 어떻게 고민하고 계신지 듣고 싶습니다. 아래에 댓글을 남겨주시거나 리포지토리(repo)에 이슈(issue)를 생성해 주세요.