9장: 보안과 거버넌스

from dataclasses import dataclass
from enum import Enum
 
class Permission(Enum):
    READ = "read"
    WRITE = "write"
    DELETE = "delete"
    EXECUTE = "execute"
    APPROVE = "approve"
 
@dataclass
class ResourcePermission:
    resource: str       # 리소스 식별자
    permissions: set[Permission]
    conditions: dict | None = None  # 조건부 권한
 
@dataclass
class AgentPermissionSet:
    agent_id: str
    role: str
    allowed_tools: list[str]
    resource_permissions: list[ResourcePermission]
    max_token_budget: int       # 토큰 사용 한도
    max_tool_calls: int         # 도구 호출 횟수 한도
    allowed_models: list[str]   # 사용 가능한 LLM 모델
    time_restrictions: dict | None = None  # 시간 기반 제한
 
# 고객 지원 에이전트 권한 예시
cs_agent_permissions = AgentPermissionSet(
    agent_id="cs-triage-agent",
    role="customer_support_triage",
    allowed_tools=[
        "customer_lookup",
        "ticket_history",
        "knowledge_base_search",
        "draft_response",
    ],
    resource_permissions=[
        ResourcePermission(
            resource="customer_db",
            permissions={Permission.READ},
        ),
        ResourcePermission(
            resource="ticket_db",
            permissions={Permission.READ, Permission.WRITE},
            conditions={"write_fields": ["status", "category", "priority"]},
        ),
        ResourcePermission(
            resource="knowledge_base",
            permissions={Permission.READ},
        ),
    ],
    max_token_budget=100_000,
    max_tool_calls=50,
    allowed_models=["claude-sonnet", "claude-haiku"],
)

class PermissionMiddleware:
    """에이전트 행동에 대한 권한 검증"""
 
    def __init__(self, permission_store: PermissionStore):
        self.store = permission_store
 
    async def check_tool_access(
        self, agent_id: str, tool_name: str
    ) -> bool:
        """에이전트의 도구 사용 권한 확인"""
        permissions = await self.store.get(agent_id)
 
        if tool_name not in permissions.allowed_tools:
            await self.audit_logger.log(
                event="permission_denied",
                agent_id=agent_id,
                tool=tool_name,
                reason="도구가 허용 목록에 없음",
            )
            return False
 
        return True
 
    async def check_resource_access(
        self, agent_id: str, resource: str, permission: Permission
    ) -> bool:
        """리소스 접근 권한 확인"""
        permissions = await self.store.get(agent_id)
 
        for rp in permissions.resource_permissions:
            if rp.resource == resource and permission in rp.permissions:
                return True
 
        await self.audit_logger.log(
            event="permission_denied",
            agent_id=agent_id,
            resource=resource,
            permission=permission.value,
        )
        return False
 
    async def check_budget(self, agent_id: str) -> bool:
        """토큰 예산 확인"""
        permissions = await self.store.get(agent_id)
        usage = await self.usage_tracker.get_current(agent_id)
 
        if usage.total_tokens >= permissions.max_token_budget:
            await self.audit_logger.log(
                event="budget_exceeded",
                agent_id=agent_id,
                budget=permissions.max_token_budget,
                used=usage.total_tokens,
            )
            return False
 
        return True

@dataclass
class ToolPolicy:
    tool_name: str
    allowed_agents: list[str]
    rate_limit: int            # 분당 최대 호출 횟수
    requires_approval: bool     # HITL 승인 필요 여부
    allowed_parameters: dict | None  # 허용된 파라미터 범위
    blocked_parameters: dict | None  # 차단된 파라미터 값
 
class ToolAccessController:
    def __init__(self, policies: list[ToolPolicy]):
        self.policies = {p.tool_name: p for p in policies}
        self.rate_counters: dict[str, RateCounter] = {}
 
    async def authorize(
        self,
        agent_id: str,
        tool_name: str,
        params: dict,
    ) -> AuthorizationResult:
        policy = self.policies.get(tool_name)
 
        if policy is None:
            return AuthorizationResult(
                allowed=False, reason=f"도구 {tool_name}에 대한 정책 미정의"
            )
 
        # 에이전트 허용 확인
        if agent_id not in policy.allowed_agents:
            return AuthorizationResult(
                allowed=False, reason=f"에이전트 {agent_id}에게 {tool_name} 미허용"
            )
 
        # Rate Limit 확인
        counter = self.rate_counters.setdefault(
            f"{agent_id}:{tool_name}", RateCounter(policy.rate_limit)
        )
        if not counter.allow():
            return AuthorizationResult(
                allowed=False, reason=f"Rate limit 초과: {policy.rate_limit}/min"
            )
 
        # 파라미터 검증
        if policy.blocked_parameters:
            for key, blocked_values in policy.blocked_parameters.items():
                if params.get(key) in blocked_values:
                    return AuthorizationResult(
                        allowed=False,
                        reason=f"차단된 파라미터 값: {key}={params[key]}"
                    )
 
        # HITL 승인 필요 여부
        if policy.requires_approval:
            return AuthorizationResult(
                allowed=True, requires_human_approval=True
            )
 
        return AuthorizationResult(allowed=True)

class InputValidator:
    """에이전트 입력 검증"""
 
    INJECTION_PATTERNS = [
        r"ignore\s+(previous|above|all)\s+(instructions?|prompts?)",
        r"system\s*:\s*",
        r"you\s+are\s+now\s+",
        r"forget\s+(everything|all|your)",
        r"new\s+instructions?\s*:",
    ]
 
    def __init__(self):
        self.patterns = [re.compile(p, re.IGNORECASE) for p in self.INJECTION_PATTERNS]
 
    def validate(self, input_text: str) -> ValidationResult:
        # 프롬프트 인젝션 패턴 탐지
        for pattern in self.patterns:
            if pattern.search(input_text):
                return ValidationResult(
                    valid=False,
                    risk="prompt_injection",
                    detail=f"의심스러운 패턴 감지: {pattern.pattern}",
                )
 
        # 입력 길이 제한
        if len(input_text) > 50_000:
            return ValidationResult(
                valid=False,
                risk="excessive_input",
                detail=f"입력 길이 초과: {len(input_text)} > 50000",
            )
 
        return ValidationResult(valid=True)

class OutputFilter:
    """에이전트 출력에서 민감 정보 필터링"""
 
    PII_PATTERNS = {
        "주민등록번호": r"\d{6}-[1-4]\d{6}",
        "전화번호": r"01[016789]-?\d{3,4}-?\d{4}",
        "카드번호": r"\d{4}[-\s]?\d{4}[-\s]?\d{4}[-\s]?\d{4}",
        "이메일": r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}",
    }
 
    def filter(self, output: str) -> FilterResult:
        masked_output = output
        detected_pii = []
 
        for pii_type, pattern in self.PII_PATTERNS.items():
            matches = re.findall(pattern, output)
            if matches:
                detected_pii.append((pii_type, len(matches)))
                # 마스킹 처리
                masked_output = re.sub(
                    pattern,
                    lambda m: self._mask(m.group(), pii_type),
                    masked_output,
                )
 
        return FilterResult(
            original_had_pii=bool(detected_pii),
            detected_types=detected_pii,
            filtered_output=masked_output,
        )
 
    def _mask(self, value: str, pii_type: str) -> str:
        if pii_type == "이메일":
            parts = value.split("@")
            return f"{parts[0][:2]}***@{parts[1]}"
        return value[:2] + "*" * (len(value) - 4) + value[-2:]

class SecretManager:
    """에이전트 비밀 관리"""
 
    def __init__(self, backend: SecretBackend):
        self.backend = backend  # Vault, AWS Secrets Manager 등
        self.cache: dict[str, tuple[str, datetime]] = {}
        self.cache_ttl = timedelta(minutes=15)
 
    async def get_secret(self, agent_id: str, secret_name: str) -> str:
        """에이전트별 비밀 조회 (캐싱 포함)"""
        # 에이전트의 비밀 접근 권한 확인
        if not await self._authorize(agent_id, secret_name):
            raise SecretAccessDenied(
                f"에이전트 {agent_id}는 {secret_name}에 접근할 수 없습니다"
            )
 
        # 캐시 확인
        cache_key = f"{agent_id}:{secret_name}"
        if cache_key in self.cache:
            value, cached_at = self.cache[cache_key]
            if datetime.utcnow() - cached_at < self.cache_ttl:
                return value
 
        # 백엔드에서 조회
        value = await self.backend.get(secret_name)
        self.cache[cache_key] = (value, datetime.utcnow())
 
        # 접근 기록
        await self.audit_logger.log(
            event="secret_accessed",
            agent_id=agent_id,
            secret_name=secret_name,
        )
 
        return value

@dataclass
class BudgetConfig:
    daily_limit_usd: float
    monthly_limit_usd: float
    per_workflow_limit_usd: float
    alert_threshold: float = 0.80  # 80% 사용 시 알림
 
class CostController:
    def __init__(self, config: BudgetConfig):
        self.config = config
        self.usage_tracker = UsageTracker()
 
    async def pre_check(self, agent_id: str, estimated_cost: float) -> CostCheckResult:
        """작업 실행 전 비용 확인"""
        daily_usage = await self.usage_tracker.get_daily(agent_id)
        monthly_usage = await self.usage_tracker.get_monthly(agent_id)
 
        # 일일 한도 확인
        if daily_usage + estimated_cost > self.config.daily_limit_usd:
            return CostCheckResult(
                allowed=False,
                reason=f"일일 예산 한도 초과: "
                       f"현재 ${daily_usage:.2f} + 예상 ${estimated_cost:.2f} > "
                       f"한도 ${self.config.daily_limit_usd:.2f}",
            )
 
        # 월간 한도 확인
        if monthly_usage + estimated_cost > self.config.monthly_limit_usd:
            return CostCheckResult(
                allowed=False,
                reason=f"월간 예산 한도 초과",
            )
 
        # 경고 임계값 확인
        daily_ratio = (daily_usage + estimated_cost) / self.config.daily_limit_usd
        if daily_ratio >= self.config.alert_threshold:
            await self.alerter.send(
                severity="warning",
                message=f"에이전트 {agent_id}: 일일 예산 {daily_ratio:.0%} 사용",
            )
 
        return CostCheckResult(allowed=True, remaining_daily=self.config.daily_limit_usd - daily_usage)
 
    async def record_usage(self, agent_id: str, cost: float, detail: dict) -> None:
        """비용 사용 기록"""
        await self.usage_tracker.record(agent_id, cost, detail)

@dataclass
class GovernancePolicy:
    """에이전트 거버넌스 정책"""
    # 배포 정책
    deployment_approval_required: bool = True
    staging_test_required: bool = True
    rollback_plan_required: bool = True
 
    # 운영 정책
    max_concurrent_workflows: int = 100
    max_workflow_duration: timedelta = timedelta(days=7)
    mandatory_hitl_for_categories: list[str] = None  # HITL 필수 카테고리
 
    # 모니터링 정책
    alert_on_error_rate_above: float = 0.05  # 5%
    alert_on_latency_above_ms: int = 30_000
    daily_report_required: bool = True
 
    # 감사 정책
    audit_log_retention_days: int = 365
    compliance_check_frequency: str = "weekly"
 
    def __post_init__(self):
        if self.mandatory_hitl_for_categories is None:
            self.mandatory_hitl_for_categories = [
                "financial_transaction",
                "personal_data_processing",
                "contract_execution",
            ]

class RiskAssessor:
    async def assess(self, agent_config: AgentConfig) -> RiskAssessment:
        """에이전트 배포 전 위험 평가"""
        risks = []
 
        # 도구 위험도 평가
        for tool in agent_config.tools:
            if tool in HIGH_RISK_TOOLS:
                risks.append(Risk(
                    category="tool_risk",
                    level="high",
                    detail=f"고위험 도구 사용: {tool}",
                    mitigation="HITL 승인 게이트 필수",
                ))
 
        # 데이터 접근 위험도
        for resource in agent_config.resources:
            if resource.contains_pii:
                risks.append(Risk(
                    category="data_risk",
                    level="high",
                    detail=f"개인정보 포함 리소스 접근: {resource.name}",
                    mitigation="출력 필터링 및 접근 로깅 필수",
                ))
 
        # 자율성 수준 위험도
        if agent_config.autonomy_level == "autonomous":
            risks.append(Risk(
                category="autonomy_risk",
                level="medium",
                detail="완전 자율 모드 운영",
                mitigation="비용 한도, 행동 모니터링, 롤백 계획 필수",
            ))
 
        overall_level = max(r.level for r in risks) if risks else "low"
 
        return RiskAssessment(
            agent_id=agent_config.agent_id,
            risks=risks,
            overall_risk_level=overall_level,
            deployment_approved=overall_level != "critical",
            required_mitigations=[r.mitigation for r in risks],
        )

class AgentMonitor:
    """에이전트 행동 모니터링"""
 
    async def check_anomalies(self, agent_id: str) -> list[Anomaly]:
        anomalies = []
        metrics = await self.metrics_store.get_recent(agent_id, minutes=15)
 
        # 에러율 급증 감지
        if metrics.error_rate > self.thresholds.error_rate:
            anomalies.append(Anomaly(
                type="error_rate_spike",
                severity="high",
                current=metrics.error_rate,
                threshold=self.thresholds.error_rate,
            ))
 
        # 비정상적 도구 호출 패턴
        if metrics.tool_call_rate > self.thresholds.tool_call_rate * 3:
            anomalies.append(Anomaly(
                type="unusual_tool_usage",
                severity="medium",
                current=metrics.tool_call_rate,
                threshold=self.thresholds.tool_call_rate,
            ))
 
        # 응답 지연 증가
        if metrics.p99_latency_ms > self.thresholds.latency_p99:
            anomalies.append(Anomaly(
                type="latency_degradation",
                severity="medium",
                current=metrics.p99_latency_ms,
                threshold=self.thresholds.latency_p99,
            ))
 
        # 반복적 동일 에러
        if metrics.repeated_error_count > 10:
            anomalies.append(Anomaly(
                type="repeated_error",
                severity="high",
                detail=f"동일 에러 {metrics.repeated_error_count}회 반복",
            ))
 
        for anomaly in anomalies:
            await self.alerter.send(agent_id=agent_id, anomaly=anomaly)
 
        return anomalies