#!/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') if route.path != '/healthz': gate = 'access-profile-v1 in enforcement mode; development: ' + gate 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='per-invocation token provider available; host adoption required')) 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()