#!/usr/bin/env python3 """Meaningful route, bad-input and filesystem cases. No foreign imports.""" import argparse import copy import errno import hashlib import importlib.util import json import os from pathlib import Path import socket import sys import tempfile BASE = Path(__file__).resolve().parents[1] spec = importlib.util.spec_from_file_location('owned_checker', BASE/'scripts/check.py') checker = importlib.util.module_from_spec(spec) spec.loader.exec_module(checker) parser = argparse.ArgumentParser() parser.add_argument('--pinned-input') parser.add_argument('--network-denied', action='store_true') args = parser.parse_args() results = [] audit_events = [] watched_dirs = [] foreign_names = {'core', 'wb_mcp', 'ozon_mcp', 'yandex_mcp', 'avito_mcp', 'ozon_perf_mcp', 'serve'} if args.pinned_input: declared = json.loads(Path(args.pinned_input).read_text(encoding='utf-8')) watched_dirs.extend(str(Path(item['local_path']).resolve().parent) for item in declared['files']) common = Path(os.path.commonpath(watched_dirs)) if common != Path(common.anchor): watched_dirs.append(str(common)) # Watch the repo root, not just passed subdirectories. watched_dirs = sorted(set(watched_dirs)) def foreign_path(value): if not isinstance(value, str): return False name = os.path.abspath(value) return '/foreign/' in name or any(name == root or name.startswith(root + os.sep) for root in watched_dirs) class ForeignModuleBlocker: def find_spec(self, fullname, path=None, target=None): if fullname.split('.')[0] in foreign_names: audit_events.append('import-block:' + fullname) raise AssertionError('foreign import') return None sys.meta_path.insert(0, ForeignModuleBlocker()) def audit(event, values): if event in ['subprocess.Popen', 'os.system'] or event.startswith('socket.'): audit_events.append(event) raise AssertionError('forbidden side effect '+event) if ((event == 'exec' and foreign_path(values[0].co_filename)) or (event == 'import' and (values[0].split('.')[0] in foreign_names or foreign_path(values[1])))): audit_events.append(event) raise AssertionError('foreign execution') if args.network_denied: denied = False try: s = socket.socket() try: s.connect(('127.0.0.1', 9)) finally: s.close() except OSError as e: denied = e.errno in [errno.EPERM, errno.EACCES] results.append({'id':'network_sandbox_control','group':'isolation','pass':denied}) sys.addaudithook(audit) def check(id_, group, condition, detail=None): results.append({'id':id_, 'group':group, 'pass':bool(condition), 'detail':detail}) def sha(data): return hashlib.sha256(data).hexdigest() def write(path, data): content = data if isinstance(data, str) else json.dumps(data, ensure_ascii=False) path.write_text(content, encoding='utf-8') return sha(content.encode()) with tempfile.TemporaryDirectory(prefix='kometa-mcp-static-') as directory: work = Path(directory).resolve() # macOS /tmp and /var are symlink aliases. policy = {'verb_default':checker.DEFAULT,'verb_floor':checker.FLOOR,'rank':checker.RANK,'confirmations':{x:checker.confirmations(x) for x in checker.LEVELS}} base_row = {'operation_id':'sample_list','method':'POST','path':'/sample/list','host':'example.invalid','safety':'read','summary':'Synthetic list'} policy_file, rows_file = work/'route-policy.json', work/'endpoints.json' def fixture(rows=None, policy_value=None): ps = write(policy_file, policy if policy_value is None else policy_value) rs = write(rows_file, [base_row] if rows is None else rows) return {'profile':'kometa-teaching-v1','commit':'synthetic-not-upstream','files':[{'logical_path':'endpoints.json','local_path':str(rows_file),'sha256':rs},{'logical_path':'route-policy.json','local_path':str(policy_file),'sha256':ps}]} def run(manifest): path = work/'manifest.json' write(path, manifest) return checker.run(str(path)) def row(**updates): return dict(base_row, **updates) r,c = run(fixture()) check('post_read_control','routes',c==0 and r.get('records', [{}])[0].get('catalog_class')=='read' and not r['internal_semantic_flags'] and r.get('records', [{}])[0].get('raw_required_confirmations')==['confirm_write']) r,c = run(fixture([row(method='GET')])) check('supported_pass_is_bounded','routes',c==0 and r['support_status']=='supported-pass' and 'complete semantic audit' in r['not_checked']) r,c = run(fixture([row(safety='write')])) check('ordinary_write_first_only','routes',c==0 and r.get('records', [{}])[0].get('catalog_required_confirmations')==['confirm_write']) r,c = run(fixture([row(safety='destructive',path='/sample/delete')])) check('catalog_destructive_two','routes',c==checker.FINDINGS_EXIT and r.get('records', [{}])[0].get('catalog_required_confirmations')==['confirm_write','i_understand_this_modifies_data']) check('raw_post_first_preserved','routes',r.get('records', [{}])[0].get('raw_required_confirmations')==['confirm_write'] and r.get('class_mismatches', [{}])[0].get('direction')=='raw_weaker') r,c = run(fixture([row(method='DELETE',safety='read')])) check('delete_floor_is_write_not_destructive','routes',c==checker.FINDINGS_EXIT and r.get('records', [{}])[0].get('catalog_class')=='write' and r.get('records', [{}])[0].get('raw_class')=='destructive' and len(r['verb_floor_overrides'])==1) r,c = run(fixture([row(operation_id='sample_custom',path='/unknown/path')])) check('unknown_path_not_semantic_pass','routes',c==0 and not r['internal_semantic_flags'] and 'current Seller API contracts' in r['not_checked']) for id_, rows, expected in [('duplicate_id',[base_row,base_row],'duplicate_operation_id'),('invalid_id',[row(operation_id='bad id')],'invalid_operation_id'),('missing_field',[{k:v for k,v in base_row.items() if k!='safety'}],'missing_or_unknown_fields'),('extra_secret_field',[dict(base_row, token='synthetic-secret-not-real')],'missing_or_unknown_fields'),('unknown_method',[row(method='OPTIONS')],'unknown_method_or_class')]: r,c = run(fixture(rows));check(id_,'bad-input',c in [2,3] and r['diagnostic']==expected and 'synthetic-secret' not in json.dumps(r)) for id_, change, expected in [('wrong_commit',lambda m:m.update(commit='other'),'unsupported_commit'),('unknown_profile',lambda m:m.update(profile='other'),'unknown_profile'),('missing_policy',lambda m:m['files'].pop(),'incomplete_input'),('duplicate_file',lambda m:m['files'].__setitem__(1,m['files'][0]),'unknown_or_duplicate_file'),('wrong_sha',lambda m:m['files'][0].update(sha256='0'*64),'supplied_sha_mismatch')]: m=fixture();change(m);r,c=run(m);check(id_,'bad-input',c in [2,3] and r['diagnostic']==expected) m=fixture(policy_value={});r,c=run(m);check('empty_policy','bad-input',c==2) altered=copy.deepcopy(policy);altered['verb_default']['POST']='read' r,c=run(fixture(policy_value=altered));check('changed_policy','bad-input',c==3 and r['diagnostic']=='unsupported_teaching_policy') m=fixture();m['files'][0]['sha256']=write(rows_file,'[{"operation_id":"one","operation_id":"two"}]') r,c=run(m);check('duplicate_json_key','bad-input',c==2 and r['diagnostic']=='duplicate_json_key') m=fixture();m['files'][0]['sha256']=write(rows_file,'[NaN]') r,c=run(m);check('nonfinite_json','bad-input',c==2 and r['diagnostic']=='nonfinite_json') m=fixture();m['files'][0]['sha256']=write(rows_file,'not json') r,c=run(m);check('malformed_json','bad-input',c==2 and r['diagnostic']=='invalid_json') m=fixture();link=work/'symlink.json';link.symlink_to(rows_file);m['files'][0]['local_path']=str(link) r,c=run(m);check('leaf_symlink','isolation',c==2 and r['diagnostic']=='symlink_component') m=fixture();dirlink=work/'dirlink';dirlink.symlink_to(work,target_is_directory=True);m['files'][0]['local_path']=str(dirlink/'endpoints.json') r,c=run(m);check('ancestor_symlink','isolation',c==2) m=fixture();fifo=work/'input-fifo';os.mkfifo(fifo);m['files'][0]['local_path']=str(fifo) r,c=run(m);check('fifo_no_block','isolation',c==2 and r['diagnostic']=='nonregular_or_oversized_file') m=fixture();m['files'][0]['sha256']=write(rows_file,'x'*(checker.MAX_FILE+1)) r,c=run(m);check('oversized','isolation',c==2) m=fixture();write(rows_file,[base_row]);before=sha(rows_file.read_bytes());r,c=run(m) check('inputs_unchanged','isolation',before==sha(rows_file.read_bytes())) # Prove the sentinels reject exec/import origins without executing foreign code. sentinel_roots = watched_dirs or [str(work/'foreign')] for event, values in [('exec', (audit.__code__.replace(co_filename=sentinel_roots[0]+'/sentinel.py'),)), ('import', ('sentinel', sentinel_roots[0]+'/sentinel.py', None, None, None))]: rejected = False try: sys.audit(event, *values) except AssertionError: rejected = True check('sentinel_control_'+event, 'isolation', rejected) rejected = False try: sys.meta_path[0].find_spec('core.synthetic_sentinel') except AssertionError: rejected = True check('sentinel_control_meta_import', 'isolation', rejected) audit_events.clear() # Positive controls are intentional; real runs must stay empty. m=fixture();m['files'][0]['local_path']='/dev/null';r,c=run(m) check('device_rejected_before_read', 'isolation', c==2 and r.get('diagnostic')=='nonregular_or_oversized_file') r,c=run(fixture());check('informational_raw_stricter', 'routes', c==0 and r.get('class_mismatch_summary')=={'raw_weaker':0,'raw_stricter':1}) bad_version=json.loads((BASE/'profiles/pinned.json').read_text());bad_version['ast_python']=[0,0] rejected=False try: checker.parse_policy('', '', bad_version) except checker.Rejected as e: rejected=e.code=='unsupported_python_version' check('ast_python_version_diagnostic','bad-input',rejected) yaml_fixture = ('endpoints:\n- operation_id: sample_list\n method: POST\n path: /sample/list\n' ' safety: read\n summary: Synthetic list\n host: example.invalid\n') for name, text in [('tag', yaml_fixture.replace('safety: read', 'safety: !!str read')), ('alias', yaml_fixture.replace('safety: read', 'safety: *unknown')), ('duplicate', yaml_fixture.replace('safety: read', 'safety: read\n safety: write')), ('multi_document', yaml_fixture + '---\n'), ('merge', yaml_fixture + ' <<: *unknown\n')]: rejected = False try: checker.project_verified_yaml(text, 'own-synthetic.yaml') except checker.Rejected: rejected = True check('yaml_projector_direct_' + name, 'bad-input', rejected) if args.pinned_input: pinned=json.loads(Path(args.pinned_input).read_text());r,c=run(pinned) idx={x['operation_id']:x for x in r.get('records',[])} supported_python=list(sys.version_info[:2])==json.loads((BASE/'profiles/pinned.json').read_text())['ast_python'] if not supported_python: check('pinned_unsupported_python','pinned',c==3 and r.get('diagnostic')=='unsupported_python_version') else: check('pinned_complete_1024','pinned',c==checker.FINDINGS_EXIT and r.get('inventory_completeness', {}).get('records')==1024) check('pinned_warehouse_evidence','pinned',idx.get('wb_post_api_warehouses',{}).get('summary')=='Создать склад продавца' and len(r.get('internal_semantic_flags',[]))==8) check('pinned_post_destructive_20','pinned',sum(x['method']=='POST' and x['catalog_class']=='destructive' and x['raw_class']=='write' for x in idx.values())==20) check('pinned_post_list_control','pinned',idx.get('ozon_product_list',{}).get('catalog_class')=='read') check('pinned_route_first_false','pinned',r.get('routes', {}).get('write_raw', {}).get('required_confirmations')==['confirm_write'] and r.get('routes', {}).get('delete_method', {}).get('required_confirmations')==['confirm_write','i_understand_this_modifies_data']) for id_, text, logical in [('foreign_execution_sentinel_sha_gate',"raise RuntimeError('foreign code executed')\n",'core/tools.py'),('dynamic_ast_policy_sha_gate',"def infer_safety():\n return __import__('os').system('false')\n",'core/safety.py'),('yaml_tag_sha_gate',"endpoints: !!python/object/apply:os.system [false]\n",'wb_mcp/endpoints.yaml'),('yaml_alias_sha_gate',"endpoints: &base\n- operation_id: *base\n",'wb_mcp/endpoints.yaml'),('yaml_duplicate_sha_gate',"endpoints:\n- operation_id: one\n safety: read\n safety: destructive\n",'wb_mcp/endpoints.yaml')]: m=copy.deepcopy(pinned); foreign=work/'foreign';foreign.mkdir(exist_ok=True);f=foreign/logical.split('/')[-1];fsha=write(f,text) for item in m['files']: if item['logical_path']==logical:item.update(local_path=str(f),sha256=fsha) r,c=run(m);check(id_,'isolation',c==3 and r['diagnostic']=='pinned_sha_mismatch') # Exercise the AST rejection path itself, with no change to trusted pins. pf=json.loads((BASE/'profiles/pinned.json').read_text()) supplied={x['logical_path']:Path(x['local_path']).read_text() for x in pinned['files']} altered=supplied['core/safety.py'].replace('return declared', 'return str(declared)') rejected=False try:checker.parse_policy(altered,supplied['core/tools.py'],pf) except checker.Rejected as e:rejected=e.code=='dynamic_or_changed_policy_ast' check('ast_fingerprint_reject','isolation',rejected) check('no_network_subprocess_foreign_exec','isolation',not audit_events, audit_events) output={'python':sys.version.split()[0],'platform':sys.platform,'network_sandbox_requested':args.network_denied,'checks_passed':sum(x['pass'] for x in results),'checks_total':len(results),'results':results,'provenance':'own synthetic cases; optional pinned text is NOT live API', 'foreign_guard_scope': watched_dirs, 'not_checked': ['pinned AST inventory requires Python 3.14'] if args.pinned_input and not supported_python else []} print(json.dumps(output,ensure_ascii=False,indent=2)) sys.exit(0 if all(x['pass'] for x in results) else 1)