# コード例1: 最小権限を実現するRBAC（Role-Based Access Control）
from enum import Enum, auto
from typing import Set, Dict, Optional, List
from functools import wraps
from datetime import datetime, timedelta
import logging
 
logger = logging.getLogger(__name__)
 
class Permission(Enum):
    READ_PRODUCTS = auto()
    WRITE_PRODUCTS = auto()
    READ_ORDERS = auto()
    WRITE_ORDERS = auto()
    MANAGE_USERS = auto()
    VIEW_ANALYTICS = auto()
    ADMIN_SETTINGS = auto()
    DELETE_DATA = auto()
    EXPORT_DATA = auto()
 
class Role(Enum):
    VIEWER = auto()
    EDITOR = auto()
    ORDER_MANAGER = auto()
    ADMIN = auto()
 
# 各ロールに最小限の権限を割り当てる
ROLE_PERMISSIONS: Dict[Role, Set[Permission]] = {
    Role.VIEWER: {
        Permission.READ_PRODUCTS,
    },
    Role.EDITOR: {
        Permission.READ_PRODUCTS,
        Permission.WRITE_PRODUCTS,
    },
    Role.ORDER_MANAGER: {
        Permission.READ_PRODUCTS,
        Permission.READ_ORDERS,
        Permission.WRITE_ORDERS,
    },
    Role.ADMIN: {
        Permission.READ_PRODUCTS,
        Permission.WRITE_PRODUCTS,
        Permission.READ_ORDERS,
        Permission.WRITE_ORDERS,
        Permission.MANAGE_USERS,
        Permission.VIEW_ANALYTICS,
        Permission.ADMIN_SETTINGS,
        # 注: DELETE_DATA と EXPORT_DATA は ADMIN にも
        # デフォルトでは付与しない（個別承認が必要）
    },
}
 
class User:
    def __init__(self, user_id: str, role: Role,
                 temporary_permissions: Optional[Set[Permission]] = None,
                 temporary_expiry: Optional[datetime] = None):
        self.user_id = user_id
        self.role = role
        self.temporary_permissions = temporary_permissions or set()
        self.temporary_expiry = temporary_expiry
 
    def get_effective_permissions(self) -> Set[Permission]:
        """有効な権限を計算する（ロール権限 + 一時権限）"""
        perms = ROLE_PERMISSIONS.get(self.role, set()).copy()
        # 一時権限が期限内なら追加
        if (self.temporary_permissions and self.temporary_expiry
                and datetime.now() < self.temporary_expiry):
            perms |= self.temporary_permissions
        return perms
 
def require_permission(permission: Permission):
    """権限チェックデコレータ — 完全仲介の原則も体現"""
    def decorator(func):
        @wraps(func)
        def wrapper(user: User, *args, **kwargs):
            effective_perms = user.get_effective_permissions()
            if permission not in effective_perms:
                logger.warning(
                    "権限不足: user=%s role=%s required=%s",
                    user.user_id, user.role.name, permission.name
                )
                raise PermissionError(
                    f"権限不足: {user.role.name} には "
                    f"{permission.name} がありません"
                )
            logger.info(
                "アクセス許可: user=%s action=%s",
                user.user_id, permission.name
            )
            return func(user, *args, **kwargs)
        return wrapper
    return decorator
 
@require_permission(Permission.WRITE_PRODUCTS)
def update_product(user: User, product_id: int, data: dict):
    """商品情報を更新する（EDITOR以上の権限が必要）"""
    # 商品更新ロジック
    pass
 
@require_permission(Permission.DELETE_DATA)
def delete_user_data(user: User, target_user_id: str):
    """ユーザーデータを削除する（個別承認された一時権限が必要）"""
    # データ削除ロジック — ADMINでもデフォルトでは実行不可
    pass
 
# 使用例: 一時的な権限昇格（Just-In-Time Access）
admin = User("admin-001", Role.ADMIN)
# GDPR削除リクエスト対応のため、一時的にDELETE_DATA権限を付与
admin.temporary_permissions = {Permission.DELETE_DATA}
admin.temporary_expiry = datetime.now() + timedelta(hours=1)
# 1時間後に自動的に権限が無効化される

# コード例2: AWS IAMポリシーの最小権限設計
import json
from typing import List, Optional
 
def create_minimal_iam_policy(
    bucket_name: str,
    prefix: str,
    allow_delete: bool = False,
    condition_ip_range: Optional[str] = None
) -> str:
    """
    特定のS3バケット・プレフィックスにのみアクセスできるポリシーを生成。
 
    設計方針:
    1. Resource は具体的に指定（ワイルドカード禁止）
    2. Action は必要最小限（Get/Put のみ、Delete はオプション）
    3. Condition で追加の制約を付与（IPレンジ、MFA等）
    4. 明示的 Deny で安全ネットを張る
    """
    statements = [
        {
            "Sid": "AllowListBucketWithPrefix",
            "Effect": "Allow",
            "Action": ["s3:ListBucket"],
            "Resource": f"arn:aws:s3:::{bucket_name}",
            "Condition": {
                "StringLike": {
                    "s3:prefix": [f"{prefix}/*"]
                }
            }
        },
        {
            "Sid": "AllowReadWriteWithPrefix",
            "Effect": "Allow",
            "Action": [
                "s3:GetObject",
                "s3:PutObject",
            ],
            "Resource": f"arn:aws:s3:::{bucket_name}/{prefix}/*"
        },
    ]
 
    # 削除権限は明示的に要求された場合のみ付与
    if allow_delete:
        statements.append({
            "Sid": "AllowDeleteWithPrefix",
            "Effect": "Allow",
            "Action": ["s3:DeleteObject"],
            "Resource": f"arn:aws:s3:::{bucket_name}/{prefix}/*",
            "Condition": {
                "Bool": {"aws:MultiFactorAuthPresent": "true"}
            }
        })
 
    # IPレンジ制約の追加
    if condition_ip_range:
        for stmt in statements:
            if stmt["Effect"] == "Allow":
                stmt.setdefault("Condition", {})
                stmt["Condition"]["IpAddress"] = {
                    "aws:SourceIp": condition_ip_range
                }
 
    # 明示的な拒否で安全ネットを張る
    statements.append({
        "Sid": "DenyDangerousActions",
        "Effect": "Deny",
        "Action": [
            "s3:DeleteBucket",
            "s3:PutBucketPolicy",
            "s3:PutBucketAcl",
            "s3:PutObjectAcl",
        ],
        "Resource": "*"
    })
 
    policy = {
        "Version": "2012-10-17",
        "Statement": statements
    }
    return json.dumps(policy, indent=2)
 
# 使用例
print(create_minimal_iam_policy(
    "my-app-data",
    "uploads/user-123",
    allow_delete=False,
    condition_ip_range="10.0.0.0/8"
))

# コード例3: JIT（Just-In-Time）アクセスマネージャー
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from typing import Dict, Optional, Callable
import threading
import logging
 
logger = logging.getLogger(__name__)
 
@dataclass
class AccessGrant:
    """一時的なアクセス許可"""
    user_id: str
    permission: str
    resource: str
    granted_at: datetime
    expires_at: datetime
    reason: str
    approved_by: str
    revoked: bool = False
 
class JITAccessManager:
    """Just-In-Time アクセス管理 — 最小権限の動的実装"""
 
    MAX_GRANT_DURATION = timedelta(hours=8)
 
    def __init__(self):
        self._grants: Dict[str, AccessGrant] = {}
        self._cleanup_timer: Optional[threading.Timer] = None
        self._start_cleanup_loop()
 
    def request_access(
        self,
        user_id: str,
        permission: str,
        resource: str,
        duration: timedelta,
        reason: str,
        approved_by: str,
    ) -> AccessGrant:
        """一時的なアクセスを要求する"""
        if duration > self.MAX_GRANT_DURATION:
            raise ValueError(
                f"最大許可時間は {self.MAX_GRANT_DURATION} です。"
                "長期アクセスが必要な場合はロール変更を申請してください。"
            )
 
        now = datetime.now()
        grant = AccessGrant(
            user_id=user_id,
            permission=permission,
            resource=resource,
            granted_at=now,
            expires_at=now + duration,
            reason=reason,
            approved_by=approved_by,
        )
 
        grant_id = f"{user_id}:{permission}:{resource}"
        self._grants[grant_id] = grant
 
        logger.info(
            "JIT access granted: user=%s perm=%s resource=%s "
            "expires=%s reason=%s approved_by=%s",
            user_id, permission, resource,
            grant.expires_at.isoformat(), reason, approved_by,
        )
 
        return grant
 
    def check_access(self, user_id: str, permission: str,
                     resource: str) -> bool:
        """アクセス権限を確認する"""
        grant_id = f"{user_id}:{permission}:{resource}"
        grant = self._grants.get(grant_id)
 
        if grant is None:
            return False
        if grant.revoked:
            return False
        if datetime.now() > grant.expires_at:
            grant.revoked = True
            logger.info("JIT access expired: user=%s perm=%s",
                       user_id, permission)
            return False
 
        return True
 
    def revoke_access(self, user_id: str, permission: str,
                      resource: str, reason: str = "manual_revoke"):
        """アクセスを即座に取り消す"""
        grant_id = f"{user_id}:{permission}:{resource}"
        grant = self._grants.get(grant_id)
        if grant and not grant.revoked:
            grant.revoked = True
            logger.info(
                "JIT access revoked: user=%s perm=%s reason=%s",
                user_id, permission, reason,
            )
 
    def _cleanup_expired(self):
        """期限切れのグラントをクリーンアップする"""
        now = datetime.now()
        expired = [
            gid for gid, g in self._grants.items()
            if now > g.expires_at or g.revoked
        ]
        for gid in expired:
            del self._grants[gid]
 
    def _start_cleanup_loop(self):
        """定期的なクリーンアップループ"""
        self._cleanup_expired()
        self._cleanup_timer = threading.Timer(60.0, self._start_cleanup_loop)
        self._cleanup_timer.daemon = True
        self._cleanup_timer.start()

# コード例4: Kubernetes RBAC — 最小権限の適用
apiVersion: rbac.authorization.k8s.io/v1
kind: Role
metadata:
  namespace: production
  name: app-deployer
rules:
  # Deployment の読み取りと更新のみ許可
  - apiGroups: ["apps"]
    resources: ["deployments"]
    verbs: ["get", "list", "watch", "update", "patch"]
    # delete は含めない（誤削除防止）
  # Pod のログ閲覧のみ許可
  - apiGroups: [""]
    resources: ["pods", "pods/log"]
    verbs: ["get", "list"]
    # exec は含めない（本番コンテナ内操作を防止）
  # ConfigMap の読み取りのみ
  - apiGroups: [""]
    resources: ["configmaps"]
    verbs: ["get", "list"]
    # Secrets へのアクセスは別途承認が必要
---
# NG: 過剰な権限付与（クラスタ管理者権限）
# apiVersion: rbac.authorization.k8s.io/v1
# kind: ClusterRoleBinding
# metadata:
#   name: dev-cluster-admin
# roleRef:
#   apiGroup: rbac.authorization.k8s.io
#   kind: ClusterRole
#   name: cluster-admin  # ← 絶対に開発者に付与しない

# コード例5: 多層防御の実装パターン — Chain of Responsibilityパターン
from typing import List, Callable, Optional, Dict, Any
from dataclasses import dataclass, field
from datetime import datetime
from abc import ABC, abstractmethod
import re
import hashlib
import logging
 
logger = logging.getLogger(__name__)
 
@dataclass
class SecurityCheckResult:
    passed: bool
    layer: str
    message: str
    severity: str = "info"  # info, warning, critical
    details: Dict[str, Any] = field(default_factory=dict)
 
@dataclass
class SecurityRequest:
    """セキュリティ検証対象のリクエスト"""
    ip: str
    method: str
    path: str
    headers: Dict[str, str]
    body: Any
    user_agent: str
    timestamp: datetime = field(default_factory=datetime.now)
 
class SecurityLayer(ABC):
    """防御層の抽象基底クラス"""
 
    @abstractmethod
    def check(self, request: SecurityRequest) -> SecurityCheckResult:
        pass
 
    @property
    @abstractmethod
    def name(self) -> str:
        pass
 
class RateLimitLayer(SecurityLayer):
    """Layer 1: レートリミット"""
    name = "RateLimit"
 
    def __init__(self, max_requests: int = 100, window_seconds: int = 60):
        self.max_requests = max_requests
        self.window_seconds = window_seconds
        self._counters: Dict[str, List[datetime]] = {}
 
    def check(self, request: SecurityRequest) -> SecurityCheckResult:
        ip = request.ip
        now = datetime.now()
        # ウィンドウ外のリクエストを除去
        self._counters.setdefault(ip, [])
        self._counters[ip] = [
            t for t in self._counters[ip]
            if (now - t).total_seconds() < self.window_seconds
        ]
        self._counters[ip].append(now)
 
        count = len(self._counters[ip])
        if count > self.max_requests:
            return SecurityCheckResult(
                False, self.name,
                f"Rate limit exceeded: {count}/{self.max_requests}",
                severity="warning",
                details={"ip": ip, "count": count}
            )
        return SecurityCheckResult(True, self.name, "Rate limit OK")
 
class WAFLayer(SecurityLayer):
    """Layer 2: WAF（Web Application Firewall）"""
    name = "WAF"
 
    DANGEROUS_PATTERNS = [
        (r"('|\")\s*(OR|AND)\s+.*=", "SQL Injection"),
        (r";\s*(DROP|DELETE|UPDATE|INSERT)", "SQL Injection"),
        (r"UNION\s+SELECT", "SQL Injection"),
        (r"<script[^>]*>", "XSS"),
        (r"javascript:", "XSS"),
        (r"on\w+\s*=", "XSS Event Handler"),
        (r"\.\./\.\./", "Path Traversal"),
        (r"/etc/passwd", "Path Traversal"),
        (r"\$\{.*\}", "Template Injection"),
        (r"{{.*}}", "Template Injection"),
    ]
 
    def check(self, request: SecurityRequest) -> SecurityCheckResult:
        body = str(request.body or "")
        path = request.path
        check_target = f"{path} {body}"
 
        for pattern, attack_type in self.DANGEROUS_PATTERNS:
            if re.search(pattern, check_target, re.IGNORECASE):
                return SecurityCheckResult(
                    False, self.name,
                    f"WAF blocked: {attack_type} detected",
                    severity="critical",
                    details={"attack_type": attack_type, "path": path}
                )
        return SecurityCheckResult(True, self.name, "WAF check passed")
 
class InputValidationLayer(SecurityLayer):
    """Layer 3: 入力値検証"""
    name = "InputValidation"
 
    MAX_FIELD_LENGTH = 10000
    MAX_BODY_SIZE = 1_000_000  # 1MB
 
    def check(self, request: SecurityRequest) -> SecurityCheckResult:
        body = request.body
        if isinstance(body, (str, bytes)):
            if len(body) > self.MAX_BODY_SIZE:
                return SecurityCheckResult(
                    False, self.name,
                    f"Body too large: {len(body)} bytes",
                    severity="warning"
                )
        elif isinstance(body, dict):
            for key, value in body.items():
                if isinstance(value, str) and len(value) > self.MAX_FIELD_LENGTH:
                    return SecurityCheckResult(
                        False, self.name,
                        f"Input too long: {key} ({len(value)} chars)",
                        severity="warning"
                    )
        return SecurityCheckResult(True, self.name, "Input validation passed")
 
class DefenseInDepth:
    """多層防御チェーンのオーケストレーター"""
 
    def __init__(self):
        self.layers: List[SecurityLayer] = []
 
    def add_layer(self, layer: SecurityLayer) -> 'DefenseInDepth':
        self.layers.append(layer)
        return self  # メソッドチェーン対応
 
    def validate(self, request: SecurityRequest) -> List[SecurityCheckResult]:
        """全防御層を順に通過させる"""
        results = []
        for layer in self.layers:
            try:
                result = layer.check(request)
                results.append(result)
                if not result.passed:
                    logger.warning(
                        "Security check failed: layer=%s msg=%s severity=%s",
                        result.layer, result.message, result.severity
                    )
                    # 失敗しても全層の状態を把握するため、
                    # critical以外は検証を続行するオプションもある
                    if result.severity == "critical":
                        break
            except Exception as e:
                result = SecurityCheckResult(
                    False, layer.name, f"Error: {e}", severity="critical"
                )
                results.append(result)
                logger.error("Security layer error: %s %s", layer.name, e)
                break
        return results
 
    def is_allowed(self, request: SecurityRequest) -> bool:
        """リクエストが全層を通過したかを判定"""
        results = self.validate(request)
        return all(r.passed for r in results)
 
# 防御チェーンの構築と使用例
defense = (DefenseInDepth()
    .add_layer(RateLimitLayer(max_requests=100))
    .add_layer(WAFLayer())
    .add_layer(InputValidationLayer()))
 
# テスト
test_request = SecurityRequest(
    ip="192.168.1.100",
    method="POST",
    path="/api/search",
    headers={"Content-Type": "application/json"},
    body={"query": "normal search term"},
    user_agent="Mozilla/5.0"
)
print(defense.is_allowed(test_request))  # True
 
malicious_request = SecurityRequest(
    ip="192.168.1.100",
    method="POST",
    path="/api/search",
    headers={"Content-Type": "application/json"},
    body={"query": "' OR 1=1 --"},
    user_agent="sqlmap/1.0"
)
print(defense.is_allowed(malicious_request))  # False

# コード例6: ゼロトラストリクエスト検証 — NIST SP 800-207準拠
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from typing import Optional, Dict, List, Tuple
from enum import Enum
import logging
 
logger = logging.getLogger(__name__)
 
class TrustLevel(Enum):
    """信頼レベル — 認証強度とコンテキストに基づく"""
    UNTRUSTED = 0
    LOW = 1
    MEDIUM = 2
    HIGH = 3
    CRITICAL = 4
 
class DeviceTrustLevel(Enum):
    UNKNOWN = "unknown"
    UNMANAGED = "unmanaged"
    MANAGED = "managed"
    COMPLIANT = "compliant"  # managedかつポリシー準拠
 
class NetworkLocation(Enum):
    PUBLIC = "public"
    VPN = "vpn"
    CORPORATE = "corporate"
 
@dataclass
class ZeroTrustContext:
    """ゼロトラスト検証に必要なコンテキスト"""
    user_id: str
    device_id: str
    device_trust_level: DeviceTrustLevel
    network_location: NetworkLocation
    mfa_verified: bool
    mfa_method: Optional[str] = None  # totp, fido2, push
    last_auth_time: datetime = field(default_factory=datetime.now)
    client_cert_valid: bool = False
    risk_score: float = 0.0          # 0.0 - 1.0
    geo_location: Optional[str] = None
    previous_geo_location: Optional[str] = None
    login_hour: Optional[int] = None
    failed_attempts: int = 0
 
@dataclass
class PolicyDecision:
    """ポリシーエンジンの判定結果"""
    allowed: bool
    trust_level: TrustLevel
    checks: Dict[str, bool]
    action_required: Optional[str] = None
    reason: Optional[str] = None
    conditions: List[str] = field(default_factory=list)
 
class ZeroTrustPolicyEngine:
    """ゼロトラストポリシーエンジン — 動的・継続的検証"""
 
    # リソースごとの要求信頼レベル
    RESOURCE_TRUST_REQUIREMENTS: Dict[str, TrustLevel] = {
        "public_api": TrustLevel.LOW,
        "user_profile": TrustLevel.MEDIUM,
        "admin_panel": TrustLevel.CRITICAL,
        "financial_data": TrustLevel.CRITICAL,
        "user_data": TrustLevel.HIGH,
        "analytics": TrustLevel.MEDIUM,
    }
 
    MAX_SESSION_AGE = timedelta(hours=1)
    MAX_RISK_SCORE = 0.7
    IMPOSSIBLE_TRAVEL_THRESHOLD_KM = 500  # 1時間以内
 
    def evaluate(self, ctx: ZeroTrustContext,
                 resource: str, action: str) -> PolicyDecision:
        """アクセスリクエストを全方位的に評価する"""
        checks = {
            "identity_verified": self._check_identity(ctx),
            "device_compliant": self._check_device(ctx),
            "network_allowed": self._check_network(ctx, resource),
            "mfa_satisfied": self._check_mfa(ctx, resource),
            "session_fresh": self._check_session_age(ctx),
            "risk_acceptable": self._check_risk(ctx),
            "no_impossible_travel": self._check_impossible_travel(ctx),
            "within_business_hours": self._check_business_hours(ctx, resource),
        }
 
        # 信頼レベルの計算
        trust_level = self._calculate_trust_level(ctx, checks)
 
        # リソースの要求レベルとの比較
        required_level = self.RESOURCE_TRUST_REQUIREMENTS.get(
            resource, TrustLevel.MEDIUM
        )
 
        all_passed = (
            all(checks.values())
            and trust_level.value >= required_level.value
        )
 
        decision = PolicyDecision(
            allowed=all_passed,
            trust_level=trust_level,
            checks=checks,
        )
 
        if not all_passed:
            decision.action_required = self._get_remediation(
                checks, trust_level, required_level
            )
            decision.reason = self._get_denial_reason(checks)
 
        # 監査ログの出力
        logger.info(
            "ZeroTrust decision: user=%s resource=%s action=%s "
            "allowed=%s trust=%s required=%s",
            ctx.user_id, resource, action,
            all_passed, trust_level.name, required_level.name,
        )
 
        return decision
 
    def _check_identity(self, ctx: ZeroTrustContext) -> bool:
        return bool(ctx.user_id)
 
    def _check_device(self, ctx: ZeroTrustContext) -> bool:
        return ctx.device_trust_level in (
            DeviceTrustLevel.MANAGED,
            DeviceTrustLevel.COMPLIANT,
        )
 
    def _check_network(self, ctx: ZeroTrustContext, resource: str) -> bool:
        sensitive_resources = {"admin_panel", "financial_data"}
        if resource in sensitive_resources:
            return ctx.network_location in (
                NetworkLocation.CORPORATE,
                NetworkLocation.VPN,
            )
        return True  # ゼロトラストではネットワーク位置は補助的
 
    def _check_mfa(self, ctx: ZeroTrustContext, resource: str) -> bool:
        sensitive_resources = {"admin_panel", "financial_data", "user_data"}
        if resource in sensitive_resources:
            if not ctx.mfa_verified:
                return False
            # 機密リソースではFIDO2を要求
            if resource == "financial_data":
                return ctx.mfa_method == "fido2"
        return True
 
    def _check_session_age(self, ctx: ZeroTrustContext) -> bool:
        return (datetime.now() - ctx.last_auth_time) < self.MAX_SESSION_AGE
 
    def _check_risk(self, ctx: ZeroTrustContext) -> bool:
        return ctx.risk_score < self.MAX_RISK_SCORE
 
    def _check_impossible_travel(self, ctx: ZeroTrustContext) -> bool:
        """不可能な移動（Impossible Travel）の検出"""
        if ctx.geo_location and ctx.previous_geo_location:
            if ctx.geo_location != ctx.previous_geo_location:
                # 実際にはジオコーディングで距離計算
                # ここでは異なる国の場合のみフラグ
                return False
        return True
 
    def _check_business_hours(self, ctx: ZeroTrustContext,
                              resource: str) -> bool:
        """営業時間外のアクセスチェック"""
        sensitive = {"admin_panel", "financial_data"}
        if resource in sensitive and ctx.login_hour is not None:
            return 6 <= ctx.login_hour <= 22
        return True
 
    def _calculate_trust_level(self, ctx: ZeroTrustContext,
                               checks: Dict[str, bool]) -> TrustLevel:
        """コンテキストに基づく信頼レベルの動的計算"""
        score = 0
        if checks["identity_verified"]:
            score += 1
        if checks["device_compliant"]:
            score += 1
        if checks["mfa_satisfied"]:
            score += 1
        if ctx.client_cert_valid:
            score += 1
        if ctx.risk_score < 0.3:
            score += 1
 
        if score >= 5:
            return TrustLevel.CRITICAL
        elif score >= 4:
            return TrustLevel.HIGH
        elif score >= 3:
            return TrustLevel.MEDIUM
        elif score >= 1:
            return TrustLevel.LOW
        return TrustLevel.UNTRUSTED
 
    def _get_remediation(self, checks: Dict[str, bool],
                         current: TrustLevel,
                         required: TrustLevel) -> str:
        if not checks["mfa_satisfied"]:
            return "MFA認証が必要です"
        if not checks["device_compliant"]:
            return "管理対象デバイスからアクセスしてください"
        if not checks["session_fresh"]:
            return "セッションが期限切れです。再認証してください"
        if not checks["no_impossible_travel"]:
            return "異常なアクセスパターンが検出されました。本人確認が必要です"
        if current.value < required.value:
            return f"このリソースには {required.name} レベルの信頼が必要です"
        return "アクセスが拒否されました。管理者に連絡してください"
 
    def _get_denial_reason(self, checks: Dict[str, bool]) -> str:
        failed = [k for k, v in checks.items() if not v]
        return f"Failed checks: {', '.join(failed)}"

# コード例7: セキュアバイデフォルトの設定クラス — 完全版
from dataclasses import dataclass, field
from typing import List, Optional, Set
import warnings
import logging
 
logger = logging.getLogger(__name__)
 
@dataclass
class SecureServerConfig:
    """セキュアバイデフォルトなサーバー設定
 
    設計方針:
    1. すべてのデフォルト値は「最も安全な設定」
    2. 緩和する場合は明示的なメソッド呼び出しが必要
    3. 緩和時には警告ログを出力
    4. 本番環境では一部の緩和を禁止
    """
 
    # --- デフォルトで安全な設定 ---
    # TLS設定
    tls_min_version: str = "TLSv1.3"
    tls_ciphers: List[str] = field(default_factory=lambda: [
        "TLS_AES_256_GCM_SHA384",
        "TLS_CHACHA20_POLY1305_SHA256",
    ])
    hsts_enabled: bool = True
    hsts_max_age: int = 31536000  # 1年
    hsts_include_subdomains: bool = True
    hsts_preload: bool = True
 
    # セキュリティヘッダー（デフォルトで有効）
    content_security_policy: str = "default-src 'self'"
    x_frame_options: str = "DENY"
    x_content_type_options: str = "nosniff"
    referrer_policy: str = "strict-origin-when-cross-origin"
    permissions_policy: str = "camera=(), microphone=(), geolocation=()"
 
    # Cookie設定（デフォルトで安全）
    cookie_secure: bool = True
    cookie_httponly: bool = True
    cookie_samesite: str = "Strict"
    cookie_max_age: int = 3600  # 1時間（短め）
 
    # CORS（デフォルトで制限）
    cors_allowed_origins: List[str] = field(default_factory=list)
    cors_allow_credentials: bool = False
    cors_max_age: int = 600  # 10分
 
    # レートリミット（デフォルトで有効）
    rate_limit_enabled: bool = True
    rate_limit_requests: int = 100
    rate_limit_window_seconds: int = 60
 
    # ログ（デフォルトで有効）
    access_log_enabled: bool = True
    error_log_enabled: bool = True
    log_sensitive_data: bool = False  # 機密データのログは無効
 
    # デバッグ（デフォルトで無効）
    debug_mode: bool = False
    expose_stack_traces: bool = False
    expose_server_version: bool = False
 
    # 環境
    _environment: str = "production"
 
    def relax_for_development(self):
        """開発環境用の緩和設定（本番では呼び出し禁止）"""
        if self._environment == "production":
            raise RuntimeError(
                "本番環境でセキュリティ設定を緩和することはできません"
            )
        logger.warning(
            "セキュリティ設定を開発環境向けに緩和しています。"
            "本番環境では絶対に使用しないでください。"
        )
        self.tls_min_version = "TLSv1.2"
        self.cors_allowed_origins = ["http://localhost:3000"]
        self.cookie_secure = False  # HTTPでの開発用
        self.debug_mode = True
 
    def add_cors_origin(self, origin: str):
        """CORS許可オリジンを追加する（監査ログ付き）"""
        if origin == "*":
            raise ValueError(
                "CORS origin に '*' は許可されません。"
                "具体的なオリジンを指定してください。"
            )
        logger.info("CORS origin added: %s", origin)
        self.cors_allowed_origins.append(origin)
 
    def relax_csp(self, directive: str, value: str):
        """CSPを緩和する（警告付き）"""
        if "'unsafe-inline'" in value or "'unsafe-eval'" in value:
            warnings.warn(
                f"CSP の {directive} に unsafe ディレクティブを追加しています。"
                "XSS のリスクが増大します。",
                SecurityWarning,
            )
        logger.warning("CSP relaxed: %s = %s", directive, value)
        self.content_security_policy = (
            f"{self.content_security_policy}; {directive} {value}"
        )
 
    def to_headers(self) -> dict:
        """HTTPレスポンスヘッダーを生成する"""
        headers = {
            "X-Content-Type-Options": self.x_content_type_options,
            "X-Frame-Options": self.x_frame_options,
            "Referrer-Policy": self.referrer_policy,
            "Content-Security-Policy": self.content_security_policy,
            "Permissions-Policy": self.permissions_policy,
        }
        if self.hsts_enabled:
            hsts_value = f"max-age={self.hsts_max_age}"
            if self.hsts_include_subdomains:
                hsts_value += "; includeSubDomains"
            if self.hsts_preload:
                hsts_value += "; preload"
            headers["Strict-Transport-Security"] = hsts_value
        if not self.expose_server_version:
            headers["Server"] = ""  # サーバーバージョンを隠蔽
        return headers
 
    def validate(self) -> List[str]:
        """設定の安全性を検証し、問題点を返す"""
        issues = []
        if self.debug_mode and self._environment == "production":
            issues.append("CRITICAL: 本番環境でデバッグモードが有効です")
        if not self.hsts_enabled:
            issues.append("WARNING: HSTSが無効です")
        if not self.cookie_secure:
            issues.append("WARNING: Cookieのsecureフラグが無効です")
        if self.cors_allowed_origins == ["*"]:
            issues.append("CRITICAL: CORSが全オリジンに開放されています")
        if self.expose_stack_traces:
            issues.append("WARNING: スタックトレースが公開されています")
        if self.log_sensitive_data:
            issues.append("WARNING: 機密データのログが有効です")
        if self.tls_min_version < "TLSv1.2":
            issues.append("CRITICAL: TLS 1.2未満は安全ではありません")
        return issues
 
class SecurityWarning(UserWarning):
    """セキュリティに関する警告"""
    pass
 
# デフォルト設定は安全
config = SecureServerConfig()
print(config.debug_mode)         # False（安全）
print(config.cookie_secure)      # True（安全）
print(config.tls_min_version)    # TLSv1.3（安全）
print(config.validate())         # []（問題なし）

# フェイルセーフの実装例
def check_authorization(user_id: str, resource: str) -> bool:
    """フェイルセーフな認可チェック"""
    try:
        # 認可サービスに問い合わせ
        result = auth_service.check(user_id, resource, timeout=5)
        if result.status == "ALLOW":
            return True
        # DENY や UNKNOWN は全て拒否
        return False
    except TimeoutError:
        # タイムアウト = 安全側に倒す = 拒否
        logger.warning("Auth check timeout for user=%s", user_id)
        return False
    except Exception as e:
        # 予期しないエラー = 安全側に倒す = 拒否
        logger.error("Auth check error: %s", e)
        return False
    # ここに到達することはないが、万が一の場合も拒否
    # return False

# Break Glass（緊急時権限昇格）の実装
class BreakGlassAccess:
    """緊急時の権限昇格メカニズム"""
 
    def activate(self, user_id: str, reason: str,
                 incident_id: str) -> str:
        """緊急権限を有効化する"""
        # 1. 複数チャネルでアラート送信
        self.alert_security_team(user_id, reason, incident_id)
        self.alert_management(user_id, reason, incident_id)
 
        # 2. 全操作を詳細にログ記録
        self.enable_detailed_audit_log(user_id)
 
        # 3. 時間制限付きの権限付与（最大2時間）
        token = self.grant_emergency_access(
            user_id,
            duration=timedelta(hours=2),
            incident_id=incident_id,
        )
 
        # 4. 自動失効タイマーの設定
        self.schedule_auto_revoke(user_id, token)
 
        return token

// NG: God Mode ポリシー
{
  "Effect": "Allow",
  "Action": "*",
  "Resource": "*"
}