AIエージェントの安全設計

"""Local HTTP authorization demonstration. No LLM or production system involved."""
from copy import deepcopy
from datetime import datetime, timezone
from http.server import BaseHTTPRequestHandler, HTTPServer
import hashlib
import json
from pathlib import Path
import platform
from threading import Thread
from urllib.error import HTTPError
from urllib.request import Request, urlopen

TOKENS = {
    "demo-editor": {"subject": "alice", "department": "A", "expires": 2000, "actions": ["read", "update"]},
    "demo-reader": {"subject": "bob", "department": "A", "expires": 2000, "actions": ["read"]},
    "demo-expired": {"subject": "alice", "department": "A", "expires": 999, "actions": ["read", "update"]},
}
NOW = 1000  # Fixed logical time; expiry behavior is reproducible.
INITIAL = {"ticket-A": {"department": "A", "text": "initial A"}, "ticket-B": {"department": "B", "text": "initial B"}}


class Handler(BaseHTTPRequestHandler):
    records = deepcopy(INITIAL)
    permission_state = "available"

    def log_message(self, *_):
        pass

    def do_POST(self):
        if self.path != "/operate":
            self.reply(404, "unknown_endpoint")
            return
        try:
            payload = json.loads(self.rfile.read(int(self.headers.get("Content-Length", "0"))))
            if not isinstance(payload, dict):
                raise ValueError("object required")
        except (ValueError, TypeError):
            self.reply(400, "invalid_json")
            return
        auth = self.headers.get("Authorization", "")
        principal = TOKENS.get(auth.removeprefix("Bearer ")) if auth.startswith("Bearer ") else None
        if principal is None or principal["expires"] <= NOW:
            self.reply(401, "invalid_or_expired_token")
            return
        if self.permission_state != "available":
            self.reply(503, "permission_check_unavailable")
            return
        action = payload.get("action")
        if action not in ("read", "update"):
            self.reply(400, "unsupported_action")
            return
        target = self.records.get(payload.get("target"))
        if target is None:
            self.reply(404, "unknown_target")
            return
        # Identity and scope come only from server-managed token/record data.
        if target["department"] != principal["department"] or action not in principal["actions"]:
            self.reply(403, "outside_scope")
            return
        if action == "update":
            if not isinstance(payload.get("text"), str):
                self.reply(400, "invalid_update")
                return
            target["text"] = payload["text"]
        self.reply(200, "allowed", target)

    def reply(self, code, reason, data=None):
        body = {"reason": reason}
        if data is not None:
            body["data"] = data
        raw = json.dumps(body).encode()
        self.send_response(code)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(raw)))
        self.end_headers()
        self.wfile.write(raw)


CASES = [
    ("担当内の閲覧", "demo-editor", {"target": "ticket-A", "action": "read"}, "available", 200, False),
    ("担当内の更新", "demo-editor", {"target": "ticket-A", "action": "update", "text": "changed"}, "available", 200, True),
    ("担当外の閲覧", "demo-editor", {"target": "ticket-B", "action": "read"}, "available", 403, False),
    ("担当外の更新", "demo-editor", {"target": "ticket-B", "action": "update", "text": "changed"}, "available", 403, False),
    ("閲覧専用の更新", "demo-reader", {"target": "ticket-A", "action": "update", "text": "changed"}, "available", 403, False),
    ("期限切れの更新", "demo-expired", {"target": "ticket-A", "action": "update", "text": "changed"}, "available", 401, False),
    ("トークンなし", None, {"target": "ticket-A", "action": "read"}, "available", 401, False),
    ("不明なトークン", "demo-unknown", {"target": "ticket-A", "action": "read"}, "available", 401, False),
    ("利用者・承認済みの偽装", "demo-editor", {"target": "ticket-B", "action": "update", "text": "changed", "subject": "admin", "department": "B", "approved": True}, "available", 403, False),
    ("権限判定サービス障害", "demo-editor", {"target": "ticket-A", "action": "update", "text": "changed"}, "unavailable", 503, False),
    ("存在しない対象", "demo-editor", {"target": "ticket-X", "action": "read"}, "available", 404, False),
    ("許可しない操作", "demo-editor", {"target": "ticket-A", "action": "delete"}, "available", 400, False),
    ("不正な更新データ", "demo-editor", {"target": "ticket-A", "action": "update", "text": 12}, "available", 400, False),
]


def main():
    server = HTTPServer(("127.0.0.1", 0), Handler)
    worker = Thread(target=server.serve_forever, daemon=True)
    worker.start()
    results = []
    try:
        for name, token, payload, state, expected_status, expected_change in CASES:
            Handler.records = deepcopy(INITIAL)
            Handler.permission_state = state
            before = deepcopy(Handler.records)
            headers = {"Content-Type": "application/json"}
            if token:
                headers["Authorization"] = "Bearer " + token
            request = Request(f"http://127.0.0.1:{server.server_port}/operate", data=json.dumps(payload).encode(), headers=headers, method="POST")
            try:
                response = urlopen(request, timeout=3)
            except HTTPError as error:
                response = error
            with response:
                status = response.code
                body = json.loads(response.read())
                reason = body["reason"]
            changed = before != Handler.records
            expected_records = deepcopy(INITIAL)
            if expected_change:
                expected_records["ticket-A"]["text"] = "changed"
            output_ok = body.get("data") == expected_records["ticket-A"] if status == 200 else "data" not in body
            passed = status == expected_status and changed == expected_change and Handler.records == expected_records and output_ok
            results.append({"case": name, "expected_status": expected_status, "actual_status": status, "expected_change": expected_change, "actual_change": changed, "reason": reason, "passed": passed})
    finally:
        server.shutdown()
        server.server_close()
        worker.join()
    report = {"executed_at_utc": datetime.now(timezone.utc).isoformat(), "python": platform.python_version(), "platform": platform.system(), "source_sha256": hashlib.sha256(Path(__file__).read_bytes()).hexdigest(), "scope": "localhost HTTP mock; fixed logical time; opaque demo tokens; no LLM, real IdP, database, external API, or production access", "passed": sum(r["passed"] for r in results), "total": len(results), "results": results}
    Path(__file__).with_name("authorization-lab-results.json").write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8")
    print(json.dumps(report, ensure_ascii=True))
    return 0 if report["passed"] == report["total"] else 1


if __name__ == "__main__":
    raise SystemExit(main())