【2026年版】ゼロトラストアーキテクチャ導入ガイド：中小企業でも実現できるセキュリティモデル

# Cloudflare Access ポリシー設定例
application:
  name: "社内管理画面"
  domain: "admin.example.com"
  type: "self_hosted"

policies:
  - name: "管理者アクセス"
    decision: "allow"
    include:
      - group: "admin-team@example.com"
    require:
      - mfa: true
      - device_posture:
          - os_version:
              operator: ">="
              value: "14.0"  # macOS 14以上
          - disk_encryption: true
          - firewall: true

  - name: "一般社員アクセス"
    decision: "allow"
    include:
      - email_domain: "example.com"
    require:
      - mfa: true
      - country: ["JP"]  # 日本国内からのアクセスのみ

# FastAPIでのOIDC認証実装例
from fastapi import FastAPI, Depends, HTTPException
from fastapi.security import OAuth2AuthorizationCodeBearer
from jose import jwt, JWTError
import httpx

app = FastAPI()

oauth2_scheme = OAuth2AuthorizationCodeBearer(
    authorizationUrl="https://idp.example.com/authorize",
    tokenUrl="https://idp.example.com/token",
)

JWKS_URL = "https://idp.example.com/.well-known/jwks.json"

async def get_current_user(token: str = Depends(oauth2_scheme)):
    """OIDCトークンの検証とユーザー情報の取得"""
    try:
        # JWKSからキーを取得
        async with httpx.AsyncClient() as client:
            jwks = await client.get(JWKS_URL)
            keys = jwks.json()["keys"]

        # トークンの検証
        payload = jwt.decode(
            token,
            keys,
            algorithms=["RS256"],
            audience="api.example.com",
            issuer="https://idp.example.com",
        )

        user_id = payload.get("sub")
        roles = payload.get("roles", [])

        if not user_id:
            raise HTTPException(status_code=401, detail="Invalid token")

        return {"user_id": user_id, "roles": roles}

    except JWTError:
        raise HTTPException(status_code=401, detail="Token validation failed")

def require_role(required_role: str):
    """ロールベースのアクセス制御デコレータ"""
    async def role_checker(user=Depends(get_current_user)):
        if required_role not in user["roles"]:
            raise HTTPException(
                status_code=403,
                detail=f"Role '{required_role}' required",
            )
        return user
    return role_checker

@app.get("/admin/dashboard")
async def admin_dashboard(user=Depends(require_role("admin"))):
    return {"message": "管理画面", "user": user}

from dataclasses import dataclass
from enum import Enum
import geoip2.database

class RiskLevel(Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"

@dataclass
class AccessContext:
    user_id: str
    device_id: str
    ip_address: str
    user_agent: str
    timestamp: float
    geo_location: str
    device_compliant: bool
    mfa_verified: bool

class ContextBasedAccessControl:
    """コンテキストベースのアクセス制御エンジン"""

    def evaluate_risk(self, context: AccessContext) -> RiskLevel:
        risk_score = 0

        # デバイスコンプライアンスチェック
        if not context.device_compliant:
            risk_score += 30

        # MFA未検証
        if not context.mfa_verified:
            risk_score += 25

        # 地理的な異常
        if context.geo_location not in ["JP", "US"]:
            risk_score += 20

        # 時間帯の異常（深夜帯）
        hour = int(context.timestamp % 86400 / 3600)
        if hour < 6 or hour > 22:
            risk_score += 15

        # リスクレベルの判定
        if risk_score >= 60:
            return RiskLevel.CRITICAL
        elif risk_score >= 40:
            return RiskLevel.HIGH
        elif risk_score >= 20:
            return RiskLevel.MEDIUM
        else:
            return RiskLevel.LOW

    def make_decision(self, risk: RiskLevel, resource_sensitivity: str) -> str:
        """リスクとリソースの機密度に基づくアクセス判定"""
        policy_matrix = {
            ("low", "public"): "allow",
            ("low", "internal"): "allow",
            ("low", "confidential"): "allow",
            ("medium", "public"): "allow",
            ("medium", "internal"): "allow_with_logging",
            ("medium", "confidential"): "step_up_auth",
            ("high", "public"): "allow_with_logging",
            ("high", "internal"): "step_up_auth",
            ("high", "confidential"): "deny",
            ("critical", "public"): "step_up_auth",
            ("critical", "internal"): "deny",
            ("critical", "confidential"): "deny",
        }
        return policy_matrix.get(
            (risk.value, resource_sensitivity), "deny"
        )