bmc_hub/app/core/economic_write_policy.py

"""The deliberately narrow set of external e-conomic mutations BMC Hub permits."""
from contextlib import contextmanager
from contextvars import ContextVar
import re

from fastapi import HTTPException

# Reading any endpoint is allowed.  Writes are an explicit allow-list, not a
# deny-list, so a new endpoint cannot accidentally become writable.
ALLOWED_WRITES = {
    ('POST', 'customers'),
    ('POST', 'products'),
    ('POST', 'orders/drafts'),
}

_approved_four_eyes_write = ContextVar('approved_four_eyes_write', default=False)
_approved_order_draft_update = ContextVar('approved_order_draft_update', default=False)
_FOUR_EYES_UPDATES = re.compile(r'^(?:customers|products)/[^/]+$')
_ORDER_DRAFT_UPDATE = re.compile(r'^orders/drafts/[0-9]+$')


@contextmanager
def approved_four_eyes_write():
    """Permit one approved customer/product update in the current async context."""
    token = _approved_four_eyes_write.set(True)
    try:
        yield
    finally:
        _approved_four_eyes_write.reset(token)


@contextmanager
def approved_order_draft_update():
    """Permit one user-confirmed update of an existing e-conomic order draft."""
    token = _approved_order_draft_update.set(True)
    try:
        yield
    finally:
        _approved_order_draft_update.reset(token)


def assert_economic_write_allowed(method: str, path: str) -> None:
    clean_path = path.split('?', 1)[0].strip('/')
    operation = (method.upper(), clean_path)
    approved_update = (
        operation[0] == 'PUT'
        and _FOUR_EYES_UPDATES.fullmatch(clean_path)
        and _approved_four_eyes_write.get()
    )
    approved_draft_update = (
        operation[0] == 'PUT'
        and _ORDER_DRAFT_UPDATE.fullmatch(clean_path)
        and _approved_order_draft_update.get()
    )
    if operation not in ALLOWED_WRITES and not approved_update and not approved_draft_update:
        raise HTTPException(
            status_code=403,
            detail='e-conomic er låst: Hub må kun oprette kunder, varer og ordrekladder',
        )