informed-decision/informed_decision/lifecycle_http.py

"""Version-one JSON adapter; only explicitly composed bearer authentication.

No listener or default authenticator. Runtime must supply an admitted identity
adapter returning Caller; browser cookies are deliberately not credentials here.
"""

from http import HTTPStatus
import json
import re
from urllib.parse import parse_qs

from .lifecycle_service import Caller, LifecycleError
from .lifecycle_commands import decode_memo
from .oidc import ORIGIN
from .records import dumps
from .store import Conflict, EvidenceUnavailable

MAX_BODY = 256 * 1024
ROUTE = re.compile(r"/v1/decisions/([A-Za-z0-9._:-]{1,256})(?:/(events|comments|sources|evaluations|revisions|publications))?")


def _object(pairs):
    result = {}
    for key, value in pairs:
        if key in result:
            raise ValueError("duplicate field")
        result[key] = value
    return result


class LifecycleAPI:
    def __init__(self, service, authenticate):
        self.service, self.authenticate = service, authenticate

    def __call__(self, environ, start_response):
        try:
            status, result = self.dispatch(environ)
        except LifecycleError as exc:
            status, result = exc.status, {"error": exc.code}
        except Conflict:
            status, result = 409, {"error": "conflict"}
        except EvidenceUnavailable:
            status, result = 404, {"error": "record_unavailable"}
        except (ValueError, TypeError, KeyError, UnicodeError, RecursionError):
            status, result = 400, {"error": "invalid_request"}
        except Exception:
            status, result = 503, {"error": "service_unavailable"}
        body = dumps(result).encode()
        start_response(f"{status} {HTTPStatus(status).phrase}", [
            ("Content-Type", "application/json; charset=utf-8"),
            ("Content-Length", str(len(body))), ("Cache-Control", "no-store"),
            ("X-Content-Type-Options", "nosniff"),
            *(([("WWW-Authenticate", "Bearer")] if status == 401 else []))])
        return [body]

    def dispatch(self, environ):
        authorization = environ.get("HTTP_AUTHORIZATION", "")
        if (not authorization.startswith("Bearer ") or not 8 <= len(authorization) <= 32775
                or any(ord(c) < 33 or ord(c) > 126 for c in authorization[7:])):
            raise LifecycleError(401, "authentication_required")
        try:
            caller = self.authenticate(authorization[7:])
            if not isinstance(caller, Caller):
                raise ValueError()
        except Exception:
            raise LifecycleError(401, "authentication_failed") from None
        self.service._caller(caller)
        path = environ.get("PATH_INFO", "")
        match = ROUTE.fullmatch(path)
        registration = path == "/v1/decisions"
        if not match and not registration:
            raise LifecycleError(404, "route_not_found")
        memo_id, resource = match.groups() if match else (None, "register")
        method = environ.get("REQUEST_METHOD", "GET")
        query = environ.get("QUERY_STRING", "")
        if len(query) > 2048:
            raise ValueError()
        params = parse_qs(query, keep_blank_values=True, strict_parsing=True, max_num_fields=4)
        if any(len(value) != 1 for value in params.values()):
            raise ValueError()
        if method == "GET" and registration:
            if set(params) - {"q", "cursor"}:
                raise ValueError()
            return 200, self.service.search(caller, query=params.get("q", [""])[0],
                                           cursor=params.get("cursor", [None])[0])
        if method == "GET" and resource is None:
            if set(params) - {"version"}:
                raise ValueError()
            version = int(params["version"][0]) if "version" in params else None
            if version is not None and version < 1:
                raise ValueError()
            return 200, self.service.retrieve(caller, memo_id, version)
        if method == "GET" and resource == "events":
            if set(params) - {"after", "limit"}:
                raise ValueError()
            return 200, {"events": self.service.events(caller, memo_id,
                after=int(params.get("after", ["0"])[0]), limit=int(params.get("limit", ["100"])[0]))}
        kinds = {"comments": "comment", "sources": "source", "evaluations": "evaluation"}
        if method != "POST" or resource not in {*kinds, "register", "revisions", "publications"}:
            raise LifecycleError(405, "method_not_allowed")
        if params:
            raise ValueError()
        if environ.get("CONTENT_TYPE", "").lower() != "application/json":
            raise LifecycleError(415, "json_required")
        length = int(environ.get("CONTENT_LENGTH", ""))
        if length < 1:
            raise ValueError()
        if length > MAX_BODY:
            raise LifecycleError(413, "body_too_large")
        raw = environ["wsgi.input"].read(length)
        if len(raw) != length:
            raise ValueError()
        data = json.loads(raw, object_pairs_hook=_object,
                          parse_constant=lambda value: (_ for _ in ()).throw(ValueError()))
        key = environ.get("HTTP_IDEMPOTENCY_KEY", "")
        if not isinstance(key, str) or not key.strip() or len(key) > 256:
            raise ValueError()
        if resource == "publications":
            if not isinstance(data, dict) or set(data) != {"version", "audience"}:
                raise ValueError()
            event_id = self.service.publish(caller, memo_id, data["version"], audience=data["audience"], key=key)
            return 200, {"event_id": event_id, "memo_id": memo_id, "version": data["version"]}
        if resource in {"register", "revisions"}:
            required = {"memo"} if registration else {"memo", "expected_version", "summary"}
            if (not isinstance(data, dict) or not required <= data.keys()
                    or data.keys() - (required | {"documents"})):
                raise ValueError()
            memo = decode_memo(data["memo"])
            documents = data.get("documents", [])
            if registration:
                event_id = self.service.register(caller, memo, key=key, documents=documents)
            else:
                if memo.id != memo_id:
                    raise ValueError()
                event_id = self.service.revise(caller, memo, key=key,
                    expected_version=data["expected_version"], summary=data["summary"], documents=documents)
            return 200, {"event_id": event_id, "memo_id": memo.id, "version": memo.version,
                         "resource": f"/v1/decisions/{memo.id}?version={memo.version}",
                         "workspace_url": f"{ORIGIN}/decisions/{memo.id}?version={memo.version}"}
        if not isinstance(data, dict) or set(data) != {"version", "body"}:
            raise ValueError()
        event_id = self.service.annotate(caller, memo_id, data["version"], kind=kinds[resource],
                                        body=data["body"], key=key)
        return 200, {"event_id": event_id, "memo_id": memo_id, "version": data["version"]}