"""0.9.7: who a seam call is, in process (tier 1) as across apps.
(b) A brick calling a neighbour as its OWN service client is a service, not a person: the callee sees
auth_source "service" and the calling brick's name (identity["service"]), so a rule like sales-order's
"only inventory's service client records a delivery with source inventory" holds in one process too.
(c) An ONWARD read on a person's behalf (caller=) is authorised by the calling brick's own service grant (the
permissions its Seam declares — the machine grants its spec declares), and scoped to the person's rows: the
person's own permissions decide what they see, the service's decide only whether the read may run.
"""
import httpx
import pytest
from erp_platform import auth, seam
from erp_platform.bricks import Brick, Refused
from erp_platform.seam import Seam, register_local, unregister_local
ITEMS = {"I-1": {"owner": "alice@shop.example"}, "I-2": {"owner": "bob@shop.example"}}
CATALOG = Brick("catalog")
@CATALOG.use_case(kind="query", route="GET /items/{item_id}", permission="catalog:item:read")
def get_item(caller: dict, item_id: str) -> dict:
"""One item, if the caller may see it (catalog:item:read_all sees every owner's)."""
scope = auth.owner_scope(caller, see_all="catalog:item:read_all")
item = ITEMS.get(item_id)
if item is None or (scope is not None and item["owner"] != scope):
raise Refused(404, f"no item {item_id}")
return {"item_id": item_id, "seen_as": caller["sub"]}
ORDERS = Brick("orders")
TO_CATALOG = Seam("catalog", "CATALOG_BASE_URL", roles=["service"], permissions=["catalog:item:read"], owner="orders")
@ORDERS.use_case(kind="command", route="POST /deliveries", permission="orders:delivery:record", idempotent=True)
def record_delivery(caller: dict, item_id: str, source: str = "manual") -> dict:
"""Record a delivery; source inventory only from inventory's service client. Reads the item as the caller."""
if source == "inventory" and (caller.get("auth_source") != "service" or caller.get("service") != "inventory"):
raise Refused(403, "only inventory's service client records a delivery with source inventory")
r = TO_CATALOG.get(f"/items/{item_id}", caller=caller)
if not r:
raise Refused(r.status or 502, r.reason)
return {"item": r.data, "auth_source": caller.get("auth_source"), "service": caller.get("service")}
STOCK_TO_ORDERS = Seam("orders", "ORDERS_BASE_URL", roles=["service"], permissions=["orders:delivery:record"],
owner="inventory")
def person(email: str, permissions: list) -> dict:
return seam.gateway_headers({"sub": email, "email": email, "app_roles": ["sales"], "permissions": permissions})
@pytest.fixture(autouse=True)
def tier1():
register_local("catalog", CATALOG.routes_app())
register_local("orders", ORDERS.routes_app())
yield
unregister_local("catalog")
unregister_local("orders")
def _post_as(headers: dict, body: dict):
with httpx.Client(transport=seam._InProcessTransport(ORDERS.routes_app()), base_url="http://t") as c:
return c.post("/deliveries", json=body, headers=headers)
def test_b_a_service_client_call_in_process_is_the_calling_brick_not_a_person():
r = STOCK_TO_ORDERS.post("/deliveries", {"item_id": "I-1", "source": "inventory"}, actor="alice@shop.example")
assert r.ok, r.reason
assert (r.data["auth_source"], r.data["service"]) == ("service", "inventory")
def test_b_a_person_cannot_claim_to_be_a_service():
"""The source a person's request carries is what the gateway minted ("pool"); a header they add is stripped there,
and in process only a seam's own call says "service"."""
r = _post_as(person("alice@shop.example", ["orders:delivery:record", "catalog:item:read"]),
{"item_id": "I-1", "source": "inventory"})
assert r.status_code == 403
def test_b_the_owner_defaults_to_the_brick_package_that_declares_the_seam():
ns: dict = {"__name__": "brick_stock.infrastructure.seams", "Seam": Seam}
exec("S = Seam('orders', 'ORDERS_BASE_URL', roles=[], permissions=[])", ns)
assert ns["S"].owner == "stock"
def test_c_an_onward_read_runs_on_the_calling_bricks_grant_and_the_persons_rows():
# alice may record deliveries but holds no catalog permission: orders' own grant (catalog:item:read) authorises
# its onward read; her row scope still decides — her item yes, bob's 404.
alice = person("alice@shop.example", ["orders:delivery:record"])
r = _post_as(alice, {"item_id": "I-1"})
assert r.status_code == 200, r.text
assert r.json()["item"] == {"item_id": "I-1", "seen_as": "alice@shop.example"}
assert _post_as(alice, {"item_id": "I-2"}).status_code == 404
def test_c_the_service_grant_never_widens_the_persons_rows():
"""A service grant that includes a see-all permission does not make the person see all: scope is theirs."""
wide = Seam("catalog", "CATALOG_BASE_URL", roles=[], permissions=["catalog:item:read", "catalog:item:read_all"],
owner="orders")
caller = auth.identity_from_headers(person("alice@shop.example", []))
assert wide.get("/items/I-1", caller=caller).ok
assert wide.get("/items/I-2", caller=caller).status == 404
def test_c_across_apps_an_acting_for_identity_carries_the_service_grant_for_the_door_only():
who = auth.identity_from_headers({
**seam.gateway_headers({"sub": "orders-svc", "email": "", "app_roles": ["service"],
"permissions": [seam.ACT_FOR, "catalog:item:read", "catalog:item:read_all"]}),
seam.ACTING_FOR_HEADER: seam.acting_for({"sub": "alice@shop.example", "email": "alice@shop.example",
"roles": ["sales"]})})
assert who["sub"] == "alice@shop.example" and who["permissions"] == []
assert auth.may_run(who, "catalog:item:read") and not auth.has_permission(who, "catalog:item:read_all")
assert auth.owner_scope(who, see_all="catalog:item:read_all") == "alice@shop.example"