hub-core/tools/build_access_inventory.py

#!/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
    from hub_core.security.browser import create_browser_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()), *routes(create_browser_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
        browser = module.endswith("security.browser")
        if browser:
            gate = "OIDC state/PKCE callback or server-side session; explicit browser composition only"
        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 'browser-session' if browser else 'hub-api'),
                             current_gate=gate, source=module,
                             conditional=browser or 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()