132 lines
6.9 KiB
Python
132 lines
6.9 KiB
Python
|
#!/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()
|