Application and API security

"""Synthetic, in-memory authorization study. Python 3.10+, standard library only.
No network/server/real credentials. Provisioning, clock, and observer are test-only.
Run: python authorization_lab.py --evidence evidence
"""
import argparse
import hashlib
import json
import platform
from copy import deepcopy
from datetime import datetime, timezone
from pathlib import Path

GRANTS = {"reader": {"read", "export"}, "editor": {"read", "write", "export"},
          "admin": {"read", "write", "export", "archive"}}
DENIED = {"status": 403, "body": {"error": "not authorized"}}

def vulnerable(s, token, action, obj):
    if not token or not obj or s.now >= token["expires"]:
        return False
    return action == "archive" or action in GRANTS.get(token["issued_role"], set())

def corrected(s, token, action, obj):
    if not token or not obj or not token["active"] or s.now >= token["expires"]:
        return False
    if token["tenant"] != obj["tenant"]:
        return False
    role = s.members.get((token["tenant"], token["subject"]))
    return action in GRANTS.get(role, set()) and (role == "admin" or token["subject"] in obj["grants"])

class Service:
    def __init__(self, policy):
        self.policy, self.now = policy, 0
        self.members = {("A", "alice"): "reader", ("A", "alex"): "editor",
                        ("A", "admin_a"): "admin", ("B", "bea"): "editor", ("B", "admin_b"): "admin"}
        self.tokens = {u: {"subject": u, "tenant": t, "issued_role": r, "active": True, "expires": 10}
                       for (t, u), r in self.members.items()}
        self.objects = {"a-shared": {"tenant": "A", "grants": {"alice", "alex"}, "text": "alpha", "archived": False},
                        "a-private": {"tenant": "A", "grants": {"alex"}, "text": "private alpha", "archived": False},
                        "b-shared": {"tenant": "B", "grants": {"bea"}, "text": "beta", "archived": False}}
        self.jobs, self.artifacts = {}, {}

    def request(self, actor, action, target):
        token, obj = self.tokens.get(actor), self.objects.get(target)
        if not self.policy(self, token, action, obj):
            return deepcopy(DENIED)
        if action == "read":
            return {"status": 200, "body": {"text": obj["text"], "archived": obj["archived"]}}
        if action == "export":
            job = f"job-{len(self.jobs) + 1}"
            self.jobs[job] = {"actor": actor, "target": target, "state": "queued"}
            return {"status": 202, "body": {"job": job}}
        if action == "write":
            obj["text"] = "updated synthetic record"
        elif action == "archive":
            obj["archived"] = True
        else:
            return deepcopy(DENIED)
        return {"status": 204, "body": {}}

    def execute(self, job_id):
        job = self.jobs[job_id]  # Trusted worker entry, supplied an existing queued job.
        token, obj = self.tokens[job["actor"]], self.objects[job["target"]]
        if job["state"] != "queued":
            return deepcopy(DENIED)
        if self.policy is corrected and not self.policy(self, token, "export", obj):
            job["state"] = "denied"
            return deepcopy(DENIED)
        self.artifacts[job_id], job["state"] = obj["text"], "ready"
        return {"status": 200, "body": {"state": "ready"}}

    def download(self, actor, job_id):
        job, token = self.jobs.get(job_id), self.tokens.get(actor)
        if not job or job_id not in self.artifacts:
            return deepcopy(DENIED)
        if not self.policy(self, token, "read", self.objects[job["target"]]):
            return deepcopy(DENIED)
        if self.policy is corrected and actor != job["actor"]:
            return deepcopy(DENIED)
        return {"status": 200, "body": {"text": self.artifacts[job_id]}}

    def change(self, event):  # Trusted fixture control, not an API endpoint.
        if event == "revoke": self.tokens["alice"]["active"] = False
        if event == "revoke_editor": self.tokens["alex"]["active"] = False
        if event == "remove": self.members.pop(("A", "alice"))
        if event == "downgrade": self.members[("A", "alex")] = "reader"
        if event == "expire": self.now = 10
        if event == "before_expiry": self.now = 9

    def observe(self):  # Test-only observer of effects, not policy internals.
        return deepcopy({"objects": self.objects, "jobs": self.jobs, "artifacts": self.artifacts})

def check(condition, detail):
    if not condition: raise AssertionError(detail)

# name, caller, operation, object, expected status, preceding fixture event
CASES = [
    ("reader_read", "alice", "read", "a-shared", 200, ""),
    ("editor_write", "alex", "write", "a-shared", 204, ""),
    ("admin_archive", "admin_a", "archive", "a-private", 204, ""),
    ("cross_tenant_read", "bea", "read", "a-shared", 403, ""),
    ("cross_tenant_write", "bea", "write", "a-shared", 403, ""),
    ("cross_tenant_admin", "admin_b", "archive", "a-shared", 403, ""),
    ("cross_tenant_export", "bea", "export", "a-shared", 403, ""),
    ("same_tenant_private", "alice", "read", "a-private", 403, ""),
    ("reader_write", "alice", "write", "a-shared", 403, ""),
    ("reader_archive", "alice", "archive", "a-shared", 403, ""),
    ("revoked_write", "alex", "write", "a-shared", 403, "revoke_editor"),
    ("removed_membership", "alice", "read", "a-shared", 403, "remove"),
    ("downgraded_write", "alex", "write", "a-shared", 403, "downgrade"),
    ("downgraded_read", "alex", "read", "a-shared", 200, "downgrade"),
    ("unknown_credential", "unknown", "read", "a-shared", 403, ""),
    ("unknown_action", "admin_a", "publish", "a-shared", 403, ""),
    ("before_expiry", "alice", "read", "a-shared", 200, "before_expiry"),
    ("at_expiry", "alice", "read", "a-shared", 403, "expire"),
    ("missing_object", "alice", "read", "missing", 403, ""),
]
FLOWS = [("export_success", "", "worker", "alice"), ("worker_removed", "remove", "worker", "alice"),
         ("worker_revoked", "revoke", "worker", "alice"), ("worker_expired", "expire", "worker", "alice"),
         ("download_revoked", "revoke", "download", "alice"), ("download_other_subject", "", "download", "alex")]
EXPECTED_FAILURES = {"cross_tenant_read", "cross_tenant_write", "cross_tenant_admin", "cross_tenant_export",
                     "same_tenant_private", "reader_archive", "revoked_write", "removed_membership", "downgraded_write",
                     "worker_removed", "worker_revoked", "worker_expired", "download_revoked", "download_other_subject"}

def scenario(policy, case):
    name, actor, action, target, want, event = case
    s = Service(policy); s.change(event)
    before = s.observe(); response = s.request(actor, action, target)
    check(response["status"] == want, f"expected {want}, received {response}")
    if want == 403:
        check(response == DENIED and s.observe() == before, "denial leaked data or changed protected state")
    elif action == "read": check(response["body"] == {"text": "alpha", "archived": False}, "incorrect read")
    elif action == "write": check(s.request("alice", "read", target)["body"]["text"] == "updated synthetic record", "write missing")
    elif action == "archive": check(s.request("alex", "read", target)["body"]["archived"], "archive missing")

def flow(policy, case):
    name, event, phase, downloader = case
    s = Service(policy); queued = s.request("alice", "export", "a-shared")
    check(queued["status"] == 202, "authorized enqueue failed")
    job = queued["body"]["job"]
    if phase == "download": check(s.execute(job)["status"] == 200, "authorized export failed")
    s.change(event)
    if phase == "worker" and event:
        check(s.execute(job) == DENIED, "worker retained revoked or expired authority")
        state = s.observe()
        check(state["jobs"][job]["state"] == "denied" and job not in state["artifacts"], "denied worker produced output")
    elif phase == "download": check(s.download(downloader, job) == DENIED, "download crossed authority boundary")
    else:
        check(s.execute(job)["status"] == 200, "authorized worker failed")
        check(s.download("alice", job) == {"status": 200, "body": {"text": "alpha"}}, "authorized download failed")

def run(policy):
    rows = []
    for test, cases in [(scenario, CASES), (flow, FLOWS)]:
        for case in cases:
            try: test(policy, case); rows.append({"scenario": case[0], "passed": True})
            except AssertionError as error: rows.append({"scenario": case[0], "passed": False, "detail": str(error)})
    return rows

def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--evidence", type=Path)
    args = parser.parse_args()
    results = {p.__name__: run(p) for p in [vulnerable, corrected]}
    failures = {key: {r["scenario"] for r in rows if not r["passed"]} for key, rows in results.items()}
    verified = failures["vulnerable"] == EXPECTED_FAILURES and not failures["corrected"]
    for key, rows in results.items(): print(f"{key}: {len(rows)-len(failures[key])}/{len(rows)} pass; {len(failures[key])} failures")
    print("Demonstration verified." if verified else "Unexpected outcome; inspect evidence.")
    if args.evidence:
        args.evidence.mkdir(parents=True, exist_ok=True)
        receipt = {"run_at_utc": datetime.now(timezone.utc).isoformat(), "python": platform.python_version(),
                   "code_sha256": hashlib.sha256(Path(__file__).read_bytes()).hexdigest(), "network_used": False,
                   "data": "synthetic", "demonstration_verified": verified, "results": results}
        (args.evidence / "run.json").write_text(json.dumps(receipt, indent=2) + "\n", encoding="utf-8")
    return 0 if verified else 1

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