#!/usr/bin/env python3 """Owned offline checker. Foreign Python is parsed as AST, never executed. Pinned YAML support is deliberately a verified-byte scalar projection, not a YAML loader. Teaching JSON is a separate disclosed synthetic data profile. """ import argparse import ast import hashlib import errno import json import os from pathlib import Path import re import stat import sys VERSION = '0.1.1' FINDINGS_EXIT = 4 BASE = Path(__file__).resolve().parents[1] MAX_FILE = 600_000 LEVELS = ('read', 'write', 'destructive') DEFAULT = {'GET': 'read', 'HEAD': 'read', 'POST': 'write', 'PUT': 'write', 'PATCH': 'write', 'DELETE': 'destructive'} FLOOR = {'PUT': 'write', 'PATCH': 'write', 'DELETE': 'write'} RANK = dict(zip(LEVELS, range(3))) NOT_CHECKED = ['runtime dispatch and alternate wrappers', 'current Seller API contracts', 'actual mutations or deletions', 'upstream tests/CI', 'credential connection safety', 'complete semantic audit', 'independent human approval', 'catalog loader normalization (core/registry.py)'] class Rejected(Exception): def __init__(self, code, status='error'): self.code, self.status = code, status def fail(code, status='error'): raise Rejected(code, status) def digest(data): return hashlib.sha256(data).hexdigest() def unique_pairs(pairs): obj = {} for k, v in pairs: if k in obj: fail('duplicate_json_key') obj[k] = v return obj def decode_json(text): try: return json.loads(text, object_pairs_hook=unique_pairs, parse_constant=lambda value: fail('nonfinite_json')) except (ValueError, RecursionError): fail('invalid_json') def keys(obj, required): if not isinstance(obj, dict) or set(obj) != set(required): fail('missing_or_unknown_fields') def read_bytes(path): """Reject symlinks in every component, devices and oversized files. Walk using held dir descriptors; no neighbour scans, resolution or imports. """ if not isinstance(path, str) or not path or '\x00' in path: fail('invalid_local_path') if not all(hasattr(os, flag) for flag in ['O_NOFOLLOW', 'O_DIRECTORY', 'O_NONBLOCK']): fail('unsupported_filesystem', 'unsupported') p = Path(path) if '..' in p.parts: fail('parent_path_not_supported') directory = os.open('/' if p.is_absolute() else '.', os.O_RDONLY | os.O_DIRECTORY) try: parts = p.parts[1:] if p.is_absolute() else p.parts if not parts: fail('invalid_local_path') for part in parts[:-1]: part_info = os.stat(part, dir_fd=directory, follow_symlinks=False) if stat.S_ISLNK(part_info.st_mode): fail('symlink_component') nxt = os.open(part, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=directory) os.close(directory) directory = nxt leaf = os.stat(parts[-1], dir_fd=directory, follow_symlinks=False) if stat.S_ISLNK(leaf.st_mode): fail('symlink_component') if not stat.S_ISREG(leaf.st_mode) or leaf.st_size > MAX_FILE: fail('nonregular_or_oversized_file') fd = os.open(parts[-1], os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK | getattr(os, 'O_NOCTTY', 0), dir_fd=directory) try: info = os.fstat(fd) if not stat.S_ISREG(info.st_mode) or info.st_size > MAX_FILE: fail('nonregular_or_oversized_file') chunks, length = [], 0 while True: data = os.read(fd, min(65536, MAX_FILE + 1 - length)) if not data: break chunks.append(data) length += len(data) if length > MAX_FILE: fail('oversized_file') return b''.join(chunks) finally: os.close(fd) except OSError as e: fail('symlink_component' if e.errno == errno.ELOOP else 'local_file_unavailable') finally: os.close(directory) def utf8(data): try: return data.decode('utf-8') except UnicodeError: fail('invalid_utf8') def valid_row(row): keys(row, ['operation_id', 'method', 'path', 'safety', 'summary', 'host']) if any(not isinstance(v, str) or not v or len(v) > 6000 for v in row.values()): fail('invalid_endpoint_field') if not re.fullmatch(r'[a-z][a-z0-9_]{1,150}', row['operation_id']): fail('invalid_operation_id') if row['method'] not in DEFAULT or row['safety'] not in LEVELS: fail('unknown_method_or_class', 'unsupported') if not row['path'].startswith('/') or any(c.isspace() for c in row['path']): fail('invalid_endpoint_path') if not re.fullmatch(r'[a-z0-9.-]+', row['host']): fail('invalid_host') def scalar(value): if not value or value[0] in '!&*|>[{' or '\n' in value: fail('unsupported_yaml_scalar', 'unsupported') if value[0] == "'": if not value.endswith("'"): fail('invalid_quoted_scalar') return value[1:-1].replace("''", "'") if value[0] == '"': result = decode_json(value) if not isinstance(result, str): fail('invalid_quoted_scalar') return result return value def project_verified_yaml(text, logical): # Called only AFTER exact pinned SHA match. No general YAML is supported. if re.search(r'^(?:---|\.\.\.)(?:\s|$)|^\s*<<\s*:', text, re.M): fail('unsupported_yaml_document_or_merge', 'unsupported') starts = list(re.finditer(r'^- operation_id: (.+)$', text, re.M)) if not starts or text[:starts[0].start()].splitlines()[-1:] != ['endpoints:']: fail('unsupported_yaml_layout', 'unsupported') rows = [] for pos, match in enumerate(starts): end = starts[pos + 1].start() if pos + 1 < len(starts) else len(text) block = text[match.end():end] top_keys = re.findall(r'^ ([a-z_]+):', block, re.M) if len(top_keys) != len(set(top_keys)): fail('duplicate_yaml_key') row = {'operation_id': scalar(match[1])} for key in ['method', 'path', 'safety', 'summary', 'host']: found = re.findall(r'^ ' + key + r': (.+(?:\n [^\n]+)*)$', block, re.M) if len(found) != 1: fail('missing_or_duplicate_yaml_field') value = found[0] if '\n' in value: if key != 'summary' or (value.startswith("'") and not value.endswith("'")): fail('unsupported_yaml_continuation', 'unsupported') value = ' '.join(x.strip() for x in value.splitlines()) row[key] = scalar(value) valid_row(row) row['evidence'] = {'file': logical, 'line': text.count('\n', 0, match.start()) + 1, 'field_lines': {key: text.count('\n', 0, match.end() + re.search(r'^ '+key+r':', block, re.M).start()) + 1 for key in ['method','path','safety','summary','host']}} rows.append(row) return rows def parse_policy(safety_text, tools_text, profile): if list(sys.version_info[:2]) != profile['ast_python']: fail('unsupported_python_version', 'unsupported') funcs = {} evidence = [] constants = {} for logical, text in [('core/safety.py', safety_text), ('core/tools.py', tools_text)]: try: tree = ast.parse(text, filename=logical) except (SyntaxError, RecursionError): fail('invalid_python_ast', 'unsupported') for n in ast.walk(tree): if isinstance(n, (ast.FunctionDef, ast.AsyncFunctionDef)) and n.name in profile['ast_fingerprints']: if n.name in funcs: fail('ambiguous_ast_function', 'unsupported') if digest(ast.dump(n, include_attributes=False).encode()) != profile['ast_fingerprints'][n.name]: fail('dynamic_or_changed_policy_ast', 'unsupported') funcs[n.name] = n evidence.append({'file': logical, 'line': n.lineno, 'function': n.name}) if logical == 'core/safety.py': for n in tree.body: if isinstance(n, ast.Assign) and len(n.targets) == 1 and isinstance(n.targets[0], ast.Name) and n.targets[0].id in ['_VERB_DEFAULT','_VERB_FLOOR','_RANK','SAFETY_LEVELS']: try: constants[n.targets[0].id] = ast.literal_eval(n.value) except (ValueError, TypeError): fail('dynamic_policy_constant', 'unsupported') if set(funcs) != set(profile['ast_fingerprints']): fail('missing_policy_function', 'unsupported') if constants != {'_VERB_DEFAULT': DEFAULT, '_VERB_FLOOR': FLOOR, '_RANK': RANK, 'SAFETY_LEVELS': LEVELS}: fail('unknown_policy_constants', 'unsupported') # Exact AST locks gate order and forwarding; these are not arbitrary Python # interpretations. Extract class labels and false-default flag signatures. routes = {} for name in ['call_method','write_method','delete_method','get_raw','write_raw','delete_raw']: fn = funcs[name] target = '_run_raw' if name.endswith('_raw') else '_run_spec' calls = [n for n in ast.walk(fn) if isinstance(n, ast.Call) and isinstance(n.func, ast.Name) and n.func.id == target] if len(calls) != 1 or not isinstance(calls[0].args[1], ast.Constant): fail('unknown_route_forwarding', 'unsupported') arguments = fn.args.args defaults = {a.arg: ast.literal_eval(d) for a, d in zip(arguments[-len(fn.args.defaults):], fn.args.defaults)} flags = [x for x in ['confirm_write','i_understand_this_modifies_data'] if defaults.get(x) is False] route_class = calls[0].args[1].value if route_class not in LEVELS or flags != confirmations(route_class): fail('unsupported_route_confirmations', 'unsupported') routes[name] = {'class': route_class, 'required_confirmations': flags, 'file': 'core/tools.py', 'line': fn.lineno} return routes, evidence def effective(method, declared): floor = FLOOR.get(method) if floor and RANK[declared] < RANK[floor]: return floor return declared def confirmations(level): return [] if level == 'read' else ['confirm_write'] if level == 'write' else ['confirm_write', 'i_understand_this_modifies_data'] def analyse(manifest): keys(manifest, ['profile', 'commit', 'files']) profile_bytes = read_bytes(str(BASE / 'profiles/pinned.json')) profile = decode_json(utf8(profile_bytes)) if manifest['profile'] not in [profile['id'], 'kometa-teaching-v1']: fail('unknown_profile', 'unsupported') teaching = manifest['profile'] == 'kometa-teaching-v1' expected_commit = 'synthetic-not-upstream' if teaching else profile['commit'] if manifest['commit'] != expected_commit: fail('unsupported_commit', 'unsupported') if not teaching and list(sys.version_info[:2]) != profile['ast_python']: fail('unsupported_python_version', 'unsupported') required_files = {'endpoints.json', 'route-policy.json'} if teaching else set(profile['files']) if not isinstance(manifest['files'], list) or len(manifest['files']) != len(required_files): fail('incomplete_input') supplied, sources = {}, [] for item in manifest['files']: keys(item, ['logical_path','local_path','sha256']) logical = item['logical_path'] if not isinstance(logical, str) or logical not in required_files or logical in supplied: fail('unknown_or_duplicate_file') if not isinstance(item['sha256'], str) or not re.fullmatch(r'[0-9a-f]{64}', item['sha256']): fail('invalid_sha256') data = read_bytes(item['local_path']) actual = digest(data) if actual != item['sha256']: fail('supplied_sha_mismatch', 'unsupported') if not teaching and actual != profile['files'][logical]: fail('pinned_sha_mismatch', 'unsupported') supplied[logical] = utf8(data) sources.append({'file': logical, 'sha256': actual, 'bytes': len(data)}) inventory, rows = {}, [] if teaching: endpoints = decode_json(supplied['endpoints.json']) if not isinstance(endpoints, list) or not 1 <= len(endpoints) <= 1500: fail('invalid_inventory') policy = decode_json(supplied['route-policy.json']) keys(policy, ['verb_default','verb_floor','rank','confirmations']) if policy != {'verb_default': DEFAULT, 'verb_floor': FLOOR, 'rank': RANK, 'confirmations': {x: confirmations(x) for x in LEVELS}}: fail('unsupported_teaching_policy', 'unsupported') for index, row in enumerate(endpoints): valid_row(row) rows.append(dict(row, evidence={'file': 'endpoints.json', 'json_pointer': '/'+str(index)})) inventory['endpoints.json'] = len(rows) routes, policy_evidence = {}, [{'file': 'route-policy.json', 'json_pointer': '/'}] semantic_ids = [] else: for logical, count in profile['inventory'].items(): part = project_verified_yaml(supplied[logical], logical) if len(part) != count: fail('incomplete_pinned_inventory') inventory[logical] = len(part) rows.extend(part) routes, policy_evidence = parse_policy(supplied['core/safety.py'], supplied['core/tools.py'], profile) semantic_ids = profile['semantic_ids'] ids = [r['operation_id'] for r in rows] if len(ids) != len(set(ids)): fail('duplicate_operation_id') records, mismatches, floors, semantics = [], [], [], [] for r in rows: catalog_class = effective(r['method'], r['safety']) raw_class = DEFAULT[r['method']] record = dict(r, declared_class=r['safety'], catalog_class=catalog_class, raw_class=raw_class, catalog_required_confirmations=confirmations(catalog_class), raw_required_confirmations=confirmations(raw_class)) records.append(record) if catalog_class != raw_class: mismatches.append({'operation_id': r['operation_id'], 'catalog_class': catalog_class, 'raw_class': raw_class, 'direction': 'raw_stricter' if RANK[raw_class]>RANK[catalog_class] else 'raw_weaker', 'evidence': r['evidence']}) if catalog_class != r['safety']: floors.append({'operation_id': r['operation_id'], 'declared_class': r['safety'], 'effective_class': catalog_class, 'evidence': r['evidence']}) if r['operation_id'] in semantic_ids: semantics.append({'operation_id': r['operation_id'], 'summary': r['summary'], 'declared_class': r['safety'], 'evidence': r['evidence'], 'status': 'pinned_manual_review_signal_not_actual_mutation'}) mismatch_summary = {direction: sum(m['direction'] == direction for m in mismatches) for direction in ('raw_weaker', 'raw_stricter')} findings = bool(mismatch_summary['raw_weaker'] or floors or semantics) return {'version': VERSION, 'support_status': 'supported-findings' if findings else 'supported-pass', 'meaning': 'Declared bounded checks completed; NEVER a server-safety recommendation', 'profile': manifest['profile'], 'profile_commit': expected_commit, 'reproducibility': {'python': sys.version.split()[0], 'ast_python': profile['ast_python'] if not teaching else None, 'profile_sha256': digest(profile_bytes), 'checker_sha256': digest(read_bytes(str(Path(__file__).resolve())))}, 'provenance': 'own synthetic teaching input; NOT upstream' if teaching else 'pinned source text; not live API', 'sources': sources, 'inventory_completeness': {'status':'complete_for_declared_profile','files':inventory,'records':len(records)}, 'policy_evidence': policy_evidence, 'routes': routes, 'records': records, 'class_mismatches': mismatches, 'class_mismatch_summary': mismatch_summary, 'class_mismatch_policy': 'raw_stricter is informational; only raw_weaker contributes findings', 'verb_floor_overrides': floors, 'internal_semantic_flags': semantics, 'not_checked': NOT_CHECKED + (['any semantic signals for synthetic input'] if teaching else [])} def run(path): try: result = analyse(decode_json(utf8(read_bytes(path)))) return result, FINDINGS_EXIT if result['support_status'] == 'supported-findings' else 0 except Rejected as e: return {'version': VERSION, 'support_status': e.status, 'diagnostic': e.code, 'inventory_completeness': {'status':'unknown'}, 'not_checked': NOT_CHECKED + ['declared checks incomplete']}, 3 if e.status == 'unsupported' else 2 except Exception: return {'version': VERSION, 'support_status':'error','diagnostic':'invalid_input_or_profile','inventory_completeness':{'status':'unknown'},'not_checked': NOT_CHECKED + ['declared checks incomplete']}, 2 def main(): parser = argparse.ArgumentParser(description='Offline bounded MCP catalogue/route-policy analysis; never executes foreign Python. No tokens or private data.') parser.add_argument('manifest', help='Explicit local UTF-8 JSON manifest with commit, exact files and SHA256; no autodownload') args = parser.parse_args() result, code = run(args.manifest) # ASCII JSON is Unicode-safe even with PYTHONIOENCODING=ascii. try: print(json.dumps(result, ensure_ascii=True, sort_keys=True, indent=2)) except Exception: return 2 # A failed stdout cannot reliably deliver a JSON error envelope. return code if __name__ == '__main__': sys.exit(main())