dodil/erp-brick-time-off

DODIL ERP brick: time off

src/brick_time_off/infrastructure/objects.py

"""The evidence store: where fit notes and other documents are kept as DataK3 objects (raw/time-off/<absence_ref>/<sha256>.<ext>).

GAP (recorded, not papered over): erp-platform 0.9.13 has no object helper, so this brick declares the PORT it needs - `EvidenceStore.put / get` -
and ships one implementation of it, DataK3's S3 endpoint (OBJECT_ENDPOINT, BUCKET, OBJECT_ACCESS_KEY, OBJECT_SECRET_KEY, OBJECT_REGION;
object.uk-lon-1.dodil.io by default), the same client expenses and gl carry. No endpoint configured is a 503 for the use case that needs it, never a
silent drop: a fit note that was not stored is not evidence.

An object is kept ONCE by key: a key that exists is never overwritten, and since the key carries the document's sha256, storing the same document
again is the same object. That is why the put may precede the row's commit: a rolled-back row leaves an orphan with its own content, never a
wrong one.
"""
import hashlib
import hmac
import os
from typing import Optional, Protocol


class StoreDown(Exception):
    """The store could not answer. `http` is what the use case answers: 503 not configured / silent, 502 failed."""

    def __init__(self, http: int, reason: str):
        super().__init__(reason)
        self.http, self.reason = http, reason


class EvidenceStore(Protocol):
    def put(self, key: str, body: bytes, content_type: str) -> str: ...
    def get(self, key: str) -> Optional[bytes]: ...


class S3Store:
    """DataK3's S3 endpoint, SigV4, over httpx. Configured from the environment."""

    def __init__(self):
        self.endpoint = os.environ.get("OBJECT_ENDPOINT", "https://object.uk-lon-1.dodil.io").rstrip("/")
        self.bucket = os.environ.get("BUCKET", "")
        self.access, self.secret = os.environ.get("OBJECT_ACCESS_KEY", ""), os.environ.get("OBJECT_SECRET_KEY", "")
        self.region = os.environ.get("OBJECT_REGION", "us-east-1")

    @property
    def configured(self) -> bool:
        return bool(self.bucket and self.access and self.secret)

    def _signed(self, method: str, key: str, body: bytes, content_type: str = "") -> tuple:
        import datetime as dt
        from urllib.parse import quote, urlparse
        host = urlparse(self.endpoint).netloc
        path = "/" + quote(f"{self.bucket}/{key}")
        now = dt.datetime.now(dt.timezone.utc)
        amz, day = now.strftime("%Y%m%dT%H%M%SZ"), now.strftime("%Y%m%d")
        payload = hashlib.sha256(body).hexdigest()
        headers = {"host": host, "x-amz-content-sha256": payload, "x-amz-date": amz}
        if content_type:
            headers["content-type"] = content_type
        signed = ";".join(sorted(headers))
        canon = "\n".join([method, path, "", "".join(f"{k}:{headers[k]}\n" for k in sorted(headers)), signed, payload])
        scope = f"{day}/{self.region}/s3/aws4_request"
        sts = "\n".join(["AWS4-HMAC-SHA256", amz, scope, hashlib.sha256(canon.encode()).hexdigest()])

        def h(k, m):
            return hmac.new(k, m.encode(), hashlib.sha256).digest()
        k = h(h(h(h(("AWS4" + self.secret).encode(), day), self.region), "s3"), "aws4_request")
        headers["authorization"] = (f"AWS4-HMAC-SHA256 Credential={self.access}/{scope}, SignedHeaders={signed}, "
                                    f"Signature={hmac.new(k, sts.encode(), hashlib.sha256).hexdigest()}")
        return f"{self.endpoint}{path}", headers

    def get(self, key: str) -> Optional[bytes]:
        import httpx
        url, headers = self._signed("GET", key, b"")
        try:
            r = httpx.get(url, headers=headers, timeout=30)
        except httpx.HTTPError as e:
            raise StoreDown(503, f"the object store did not answer ({e.__class__.__name__})") from None
        if r.status_code == 404:
            return None
        if r.status_code != 200:
            raise StoreDown(502, f"the object store answered {r.status_code}")
        return r.content

    def put(self, key: str, body: bytes, content_type: str) -> str:
        import httpx
        if self.get(key) is not None:
            return key
        url, headers = self._signed("PUT", key, body, content_type)
        try:
            r = httpx.put(url, headers=headers, content=body, timeout=30)
        except httpx.HTTPError as e:
            raise StoreDown(503, f"the object store did not answer ({e.__class__.__name__})") from None
        if r.status_code >= 300:
            raise StoreDown(502, f"the object store answered {r.status_code}")
        return key


class MemoryStore:
    """For tests and local runs: a dict."""

    def __init__(self):
        self.objects: dict = {}
        self.puts = 0
        self.down: Optional[StoreDown] = None

    def put(self, key: str, body: bytes, content_type: str) -> str:
        if self.down:
            raise self.down
        self.puts += 1
        self.objects.setdefault(key, (body, content_type))
        return key

    def get(self, key: str) -> Optional[bytes]:
        if self.down:
            raise self.down
        got = self.objects.get(key)
        return got[0] if got else None


class Ports:
    """`store` is replaceable (tests do); None means S3 from the environment, and an unconfigured environment is a 503 naming what to set."""

    def __init__(self):
        self.store: Optional[EvidenceStore] = None

    def evidence(self) -> EvidenceStore:
        if self.store is not None:
            return self.store
        s3 = S3Store()
        if not s3.configured:
            raise StoreDown(503, "no evidence store is configured: set OBJECT_ENDPOINT, BUCKET, OBJECT_ACCESS_KEY and OBJECT_SECRET_KEY")
        return s3


ports = Ports()