dodil/erp-brick-gl

DODIL ERP brick: general ledger

src/brick_gl/infrastructure/objects.py

"""Objects the books produce or keep: the statutory files (FEC, SAF-T) under raw/audit/<company>/<fy>/, close evidence under
raw/close/. One port, two implementations: DataK3's S3 endpoint (object.uk-lon-1.dodil.io, SigV4: OBJECT_ENDPOINT, BUCKET,
OBJECT_ACCESS_KEY, OBJECT_SECRET_KEY) where it is configured, else a table of this brick (gl_object). An object is kept ONCE by
key: a key that exists is never overwritten (evidence is not replaced)."""
import base64
import hashlib
import os
from typing import Optional


class Missing(Exception):
    def __init__(self, what: str, reason: str, status: int = 502):
        super().__init__(f"{what}: {reason}")
        self.what, self.reason, self.status = what, reason, status


class TableObjects:
    def __init__(self, repo):
        self.repo = repo

    def put(self, key: str, body: bytes, content_type: str) -> str:
        from brick_gl.infrastructure import tables as t
        from brick_gl.infrastructure.repositories import _Keyed
        keyed = _Keyed(self.repo.conn, t.gl_object, ("object_key",))
        if keyed.get(key) is None:
            keyed.add({"object_key": key, "content_type": content_type, "body": base64.b64encode(body).decode(), "sha256": hashlib.sha256(body).hexdigest()})
        return key

    def get(self, key: str) -> Optional[bytes]:
        from brick_gl.infrastructure import tables as t
        from brick_gl.infrastructure.repositories import _Keyed
        row = _Keyed(self.repo.conn, t.gl_object, ("object_key",)).get(key)
        return base64.b64decode(row["body"]) if row else None


class S3Objects:
    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
        import hmac
        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"")
        r = httpx.get(url, headers=headers, timeout=30)
        return r.content if r.status_code == 200 else None

    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)
        r = httpx.put(url, headers=headers, content=body, timeout=30)
        if r.status_code >= 300:
            raise Missing("object store", f"{r.status_code}: {r.text[:200]}", r.status_code)
        return key


class Ports:
    """Replace `objects` to swap the store (tests do); None = S3 from the environment, else the table."""

    def __init__(self):
        self.objects = None

    def store(self, repo):
        if self.objects is not None:
            return self.objects
        s3 = S3Objects()
        return s3 if s3.configured else TableObjects(repo)


ports = Ports()