docs: inventory Hub and platform access boundaries for root integration

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a0e715-b802-70f0-a8fa-590d9ee673a5

tools/build_access_inventory.py

@@ -0,0 +1,131 @@
+#!/usr/bin/env python3
+"""Inventory source surfaces without network calls, database access or credentials.
+
+Run with hub-core's runtime environment. --check detects source-surface drift;
+it is not an authorization conformance test. Platform snapshot is reviewed input.
+"""
+import argparse
+import ast
+import inspect
+import json
+from pathlib import Path
+import sys
+
+
+def discover(root):
+    sys.path.insert(0, str(root))
+    from hub_core.runtime.app import create_app
+    from hub_core.runtime.config import RuntimeSettings
+    from hub_core.runtime.inbox_projection import create_inbox_projection_router
+
+    rows = []
+
+    def routes(router):
+        for route in router.routes:
+            if hasattr(route, 'original_router'):
+                yield from routes(route.original_router)
+            elif hasattr(route, 'methods'):
+                yield route
+            else:
+                raise ValueError(f'Uninventoried route type: {type(route).__name__}')
+
+    # Construction only: no lifespan/startup and no requests or DB connection.
+    app = create_app(settings=RuntimeSettings())
+    for route in [*routes(app), *routes(create_inbox_projection_router())]:
+        endpoint = route.endpoint
+        source = inspect.getsource(endpoint)
+        module = endpoint.__module__
+        gate = ('shared-bearer' if '_protected(' in source or module.endswith('inbox_projection')
+                else 'no-identity-check-in-handler')
+        for method in sorted(route.methods):
+            rows.append(dict(id=f'http:{method}:{route.path}:{module}.{endpoint.__name__}', kind='runtime-http',
+                             method=method, path=route.path, profile=('minimal-health' if route.path == '/healthz' else 'hub-api'),
+                             current_gate=gate, source=module,
+                             conditional=module.endswith('inbox_projection'),
+                             handler=endpoint.__name__))
+
+    verbs = {'get', 'post', 'patch', 'put', 'delete', 'head', 'options'}
+    for path in sorted((root / 'hub_core/routers').glob('*.py')):
+        tree = ast.parse(path.read_text())
+        for factory in tree.body:
+            if not isinstance(factory, ast.FunctionDef) or not factory.name.startswith('create_'):
+                continue
+            prefix = ''
+            for node in ast.walk(factory):
+                if isinstance(node, ast.Call) and isinstance(node.func, ast.Name) and node.func.id == 'APIRouter':
+                    for kw in node.keywords:
+                        if kw.arg == 'prefix':
+                            if isinstance(kw.value, ast.Constant):
+                                prefix = kw.value.value
+                            elif isinstance(kw.value, ast.Name):
+                                defaults = dict(zip([a.arg for a in factory.args.kwonlyargs], factory.args.kw_defaults))
+                                prefix = ast.literal_eval(defaults[kw.value.id])
+                            else:
+                                raise ValueError('Unresolved SDK prefix')
+            for node in ast.walk(factory):
+                if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
+                    continue
+                for dec in node.decorator_list:
+                    if isinstance(dec, ast.Call) and isinstance(dec.func, ast.Attribute) and dec.func.attr in verbs:
+                        route_path = prefix + ast.literal_eval(dec.args[0])
+                        method = dec.func.attr.upper()
+                        rows.append(dict(id=f'sdk:{factory.name}:{method}:{route_path}',
+                                         kind='embedded-http', method=method, path=route_path,
+                                         profile='embedded-host', factory=factory.name,
+                                         source=str(path.relative_to(root)), line=node.lineno,
+                                         current_gate='host-injected; not established by inventory',
+                                         conditional=True))
+
+    path = root / 'hub_core/mcp/server.py'
+    tree = ast.parse(path.read_text())
+    expected = None
+    for node in tree.body:
+        if isinstance(node, ast.Assign) and any(isinstance(t, ast.Name) and t.id == 'CORE_TOOL_NAMES' for t in node.targets):
+            expected = set(ast.literal_eval(node.value.args[0]))
+    for node in ast.walk(tree):
+        if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
+            continue
+        for dec in node.decorator_list:
+            if isinstance(dec, ast.Call) and isinstance(dec.func, ast.Name) and dec.func.id == 'register':
+                name = ast.literal_eval(dec.args[0])
+                calls = []
+                for call in ast.walk(node):
+                    if isinstance(call, ast.Call) and isinstance(call.func, ast.Attribute) and call.func.attr in {'_get','_post','_patch','_put','_delete'}:
+                        calls.append(dict(method=call.func.attr[1:].upper(), path_expression=ast.unparse(call.args[0])))
+                rows.append(dict(id=f'mcp:{name}', kind='mcp', tool=name, profile='mcp-client',
+                                 source=str(path.relative_to(root)), line=node.lineno,
+                                 target_calls=calls, current_gate='no per-user credential forwarding in base wrapper'))
+    assert expected == {r['tool'] for r in rows if r['kind'] == 'mcp'}
+    assert len(rows) == len({r['id'] for r in rows}), 'Duplicate surface identity'
+    return sorted(rows, key=lambda r:r['id'])
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('--root', type=Path, default=Path(__file__).resolve().parents[1])
+    parser.add_argument('--inventory', type=Path, required=True)
+    parser.add_argument('--check', action='store_true')
+    args = parser.parse_args()
+    data = json.loads(args.inventory.read_text())
+    found = discover(args.root)
+    if args.check:
+        assert data['hub_surfaces'] == found, 'Source inventory drift; regenerate and review'
+    else:
+        data['hub_surfaces'] = found
+        args.inventory.write_text(json.dumps(data, indent=2)+'\n')
+    cases = data['acceptance_cases']
+    profiles = data['profiles']
+    for row in data['hub_surfaces'] + data['platform_surfaces']:
+        profile = profiles[row['profile']]
+        assert all(profile.get(k) for k in ('audience','actor_tenant','target_tenant','action_resource_rule','enforcement','test_owner','cases'))
+        assert all(c in cases for c in profile['cases'])
+    objects = data['cluster_snapshot']
+    mapped = [x for r in data['platform_surfaces'] for x in r.get('objects',[])]
+    keys = lambda xs: sorted((x['kind'],x['namespace'],x['name']) for x in xs)
+    assert keys(mapped) == keys(objects), 'Cluster object coverage mismatch or duplicates'
+    assert all(r.get('owner') and r.get('review_status') for r in data['platform_surfaces'])
+    print(f"{len(found)} Hub surfaces; {len(data['platform_surfaces'])} platform rows; {len(objects)} cluster objects; inventory checks pass")
+
+
+if __name__ == '__main__':
+    main()