#!/usr/bin/env python3 """Bounded, read-only acceptance of independently captured delegation evidence.""" from __future__ import annotations import argparse, ast, hashlib, json, os, re, selectors, signal, stat, subprocess, time from typing import NoReturn class EvidenceError(ValueError): pass _HEX40=re.compile(r"^[0-9a-f]{40}$"); _HEX64=re.compile(r"^[0-9a-f]{64}$") _CONTEXT="gitea-agent.delegation-evidence-context/v1"; _RESULT="gitea-agent.delegation-evidence-result/v1" _MAX_JSON=1024*1024; _MAX_SOURCE=1024*1024; _MAX_LIST=512; _GIT="/usr/bin/git" def _bad() -> NoReturn: raise EvidenceError("rejected evidence input") def _keys(v, names): if not isinstance(v,dict) or set(v)!=set(names): _bad() def _nonempty(v): return isinstance(v,str) and bool(v.strip()) and "\x00" not in v def _path(v): if not isinstance(v,str) or not v or v.startswith("/") or "\\" in v or "\x00" in v or any(x in ("",".","..") for x in v.split("/")): _bad() return v def _digest(v, rx=_HEX64): if not isinstance(v,str) or rx.fullmatch(v) is None: _bad() return v def _bounded_list(v): if not isinstance(v,list) or len(v)>_MAX_LIST: _bad() return v def _abs(v): if not isinstance(v,str) or not os.path.isabs(v) or v!=os.path.normpath(v) or "\x00" in v: _bad() return v def _directory_fd(path): _abs(path); fd=os.open("/",os.O_RDONLY|os.O_DIRECTORY|os.O_NOFOLLOW|os.O_CLOEXEC) try: for part in path.split("/")[1:]: if part: n=os.open(part,os.O_RDONLY|os.O_DIRECTORY|os.O_NOFOLLOW|os.O_CLOEXEC,dir_fd=fd); os.close(fd); fd=n if not stat.S_ISDIR(os.fstat(fd).st_mode): _bad() return fd except (OSError,EvidenceError): try: os.close(fd) except OSError: pass _bad() def _open_rel(root_fd,path): cur=os.dup(root_fd) try: parts=path.split("/") for part in parts[:-1]: n=os.open(part,os.O_RDONLY|os.O_DIRECTORY|os.O_NOFOLLOW|os.O_CLOEXEC,dir_fd=cur); os.close(cur); cur=n fd=os.open(parts[-1],os.O_RDONLY|os.O_NOFOLLOW|os.O_CLOEXEC|os.O_NONBLOCK,dir_fd=cur) if not stat.S_ISREG(os.fstat(fd).st_mode): os.close(fd); _bad() return fd except (OSError,EvidenceError): _bad() finally: try: os.close(cur) except OSError: pass def _read(root_fd,path,limit=_MAX_SOURCE): fd=_open_rel(root_fd,path) try: data=os.read(fd,limit+1) finally: os.close(fd) if len(data)>limit: _bad() return data def _json_file(path): rawpath=_abs(path); parent=_directory_fd(os.path.dirname(rawpath)) try: fd=_open_rel(parent,os.path.basename(rawpath)) try: raw=os.read(fd,_MAX_JSON+1) finally: os.close(fd) finally: os.close(parent) if len(raw)>_MAX_JSON: _bad() def pairs(items): out={} for k,v in items: if k in out: _bad() out[k]=v return out try: return json.loads(raw.decode(),object_pairs_hook=pairs,parse_constant=lambda _: _bad()) except EvidenceError: raise except (UnicodeError,ValueError,TypeError,RecursionError): _bad() def _canon(v): return hashlib.sha256(json.dumps(v,sort_keys=True,separators=(",",":"),ensure_ascii=True).encode()).hexdigest() def _env(): return {"PATH":"/usr/bin:/bin","HOME":"/nonexistent","GIT_CONFIG_NOSYSTEM":"1","GIT_CONFIG_GLOBAL":os.devnull,"GIT_CONFIG_SYSTEM":os.devnull,"GIT_NO_LAZY_FETCH":"1","GIT_NO_REPLACE_OBJECTS":"1","GIT_TERMINAL_PROMPT":"0"} def _git(gitdir,args,fd): cmd=[_GIT,"--git-dir",gitdir,"--literal-pathspecs",*args]; p=None try: p=subprocess.Popen(cmd,env=_env(),stdin=subprocess.DEVNULL,stdout=subprocess.PIPE,stderr=subprocess.DEVNULL,start_new_session=True,pass_fds=(fd,)); out=bytearray(); deadline=time.monotonic()+5 with selectors.DefaultSelector() as s: s.register(p.stdout,selectors.EVENT_READ) while True: left=deadline-time.monotonic() if left<=0 or not s.select(left): raise subprocess.TimeoutExpired(cmd,5) chunk=os.read(p.stdout.fileno(),min(65536,_MAX_SOURCE-len(out)+1)) if not chunk: break out.extend(chunk) if len(out)>_MAX_SOURCE: raise subprocess.SubprocessError() return p.wait(timeout=max(.001,deadline-time.monotonic())),bytes(out) except (OSError,subprocess.SubprocessError): if p and p.returncode is None: try: os.killpg(p.pid,signal.SIGKILL) except ProcessLookupError: pass p.wait() return None finally: if p and p.stdout: p.stdout.close() def _context(c): _keys(c,("schema","source","stage","bundle_sha256","files","allowed_actions","obligations","executions")) if c["schema"]!=_CONTEXT or not _nonempty(c["stage"]): _bad() src=c["source"] if not isinstance(src,dict) or src.get("kind") not in ("git_tree","captured_worktree"): _bad() if src["kind"]=="git_tree": _keys(src,("kind","revision")) if not _HEX40.fullmatch(src["revision"]): _bad() else: _keys(src,("kind","base_revision")) if not _HEX40.fullmatch(src["base_revision"]): _bad() files=c["files"] if not isinstance(files,dict) or len(files)>_MAX_LIST: _bad() for p,h in files.items(): _path(p); _digest(h) if c["bundle_sha256"]!=_canon(files): _bad() actions=_bounded_list(c["allowed_actions"]) if len(set(actions))!=len(actions) or any(a not in ("static","execute") for a in actions): _bad() obs=_bounded_list(c["obligations"]); ids=set() for o in obs: _keys(o,("id","kind")) if not _nonempty(o["id"]) or o["id"] in ids or o["kind"] not in ("static","execution","red_green"): _bad() ids.add(o["id"]) exs=_bounded_list(c["executions"]); eids=set() for e in exs: _keys(e,("id","test_id","bundle_sha256","test_file_sha256","argv","exit_code","failure_kind","evidence_path","evidence_sha256","origin")) if not _nonempty(e["id"]) or e["id"] in eids or not isinstance(e["test_id"],str) or e["origin"] not in ("direct","inherited"): _bad() eids.add(e["id"]); _digest(e["bundle_sha256"]); _digest(e["test_file_sha256"]); av=_bounded_list(e["argv"]) if not av or any(not _nonempty(x) for x in av): _bad() if isinstance(e["exit_code"],bool) or not isinstance(e["exit_code"],int) or e["failure_kind"] not in ("none","assertion","approved_behavior","import","environment","timeout"): _bad() _path(e["evidence_path"]); _digest(e["evidence_sha256"]) return src,files,actions,obs,exs def _result(r,c): _keys(r,("schema","source","stage","bundle_sha256","claims")) if r["schema"]!=_RESULT or not _nonempty(r["stage"]): _bad() source=r["source"] if not isinstance(source,dict) or source.get("kind") not in ("git_tree","captured_worktree"): _bad() field="revision" if source["kind"]=="git_tree" else "base_revision" _keys(source,("kind",field)); _digest(source[field],_HEX40); _digest(r["bundle_sha256"]) claims=_bounded_list(r["claims"]); seen=set() for q in claims: _keys(q,("id","status","anchors","execution_ids")) if not _nonempty(q["id"]) or q["id"] in seen or q["status"] not in ("supported","indeterminate"): _bad() seen.add(q["id"]); _bounded_list(q["anchors"]); ids=_bounded_list(q["execution_ids"]) if len(set(ids))!=len(ids) or any(not _nonempty(i) for i in ids): _bad() return claims def _has_test(raw,symbol): if not isinstance(symbol,str) or not symbol or any(not x.isidentifier() for x in symbol.split(".")): return False try: tree=ast.parse(raw.decode()) except (SyntaxError,UnicodeError,RecursionError): return False want=tuple(symbol.split(".")); found=[] def walk(nodes,prefix=()): for n in nodes: if isinstance(n,(ast.FunctionDef,ast.AsyncFunctionDef,ast.ClassDef)): cur=prefix+(n.name,) if cur==want and isinstance(n,(ast.FunctionDef,ast.AsyncFunctionDef)) and n.name.startswith("test"): found.append(n) if isinstance(n,ast.ClassDef): walk(n.body,cur) walk(tree.body); return len(found)==1 def _git_files(src,files,git_fd,root_fd): if git_fd is None: return False,{} gd=f"/proc/self/fd/{git_fd}"; revision=src["revision"] frozen={p:_read(root_fd,p) for p in files} if any(len(frozen[p])>_MAX_SOURCE or hashlib.sha256(frozen[p]).hexdigest()!=h for p,h in files.items()): return False, frozen typ=_git(gd,["cat-file","-t",revision],git_fd) if typ is None or typ[0]!=0 or typ[1].strip()!=b"commit": return False, frozen if not files: return True, frozen tree=_git(gd,["ls-tree","-rz","-r","--full-tree",revision,"--",*sorted(files)],git_fd) if tree is None or tree[0]!=0: return False, frozen out={} try: rows=tree[1].split(b"\0") for row in rows: if not row: continue head,pathb=row.split(b"\t",1); mode,kind,obj=head.decode().split(" "); path=pathb.decode() if mode not in ("100644","100755") or kind!="blob" or path not in files: return False, frozen blob=_git(gd,["show",f"{revision}:{path}"],git_fd) if blob is None or blob[0]!=0 or len(blob[1])>_MAX_SOURCE: return False, frozen if hashlib.sha1(b"blob "+str(len(blob[1])).encode()+b"\0"+blob[1]).hexdigest()!=obj: return False, frozen if hashlib.sha256(blob[1]).hexdigest()!=files[path] or blob[1]!=frozen[path]: return False, frozen out[path]=blob[1] except (UnicodeError,ValueError): return False, frozen return set(out)==set(files), frozen def _git_has_path(src,path,git_fd): out=_git(f"/proc/self/fd/{git_fd}",["ls-tree","-rz","-r","--full-tree",src["revision"],"--",path],git_fd) if out is None or out[0]!=0: return None return bool(out[1].rstrip(b"\0")) def verify_evidence(bundle_root,context,result,git_dir=None): rootfd=_directory_fd(_abs(bundle_root)) gitfd=None try: src,files,actions,obs,exs=_context(context); claims=_result(result,context); cmap={q["id"]:q for q in claims}; omap={o["id"]:o for o in obs} if set(cmap)!=set(omap): _bad() if src["kind"]=="git_tree": if git_dir is None: ok,raw=False,{} else: try: gitfd=_directory_fd(git_dir) except EvidenceError: ok,raw=False,{} else: ok,raw=_git_files(src,files,gitfd,rootfd) else: raw={p:_read(rootfd,p) for p in files}; ok=all(hashlib.sha256(raw[p]).hexdigest()==h for p,h in files.items()) reasons=[] for field in ("source","stage","bundle_sha256"): if result[field]!=context[field]: reasons.append("result_"+field+"_mismatch") if not ok: reasons.append("source_hash_mismatch") emap={e["id"]:e for e in exs}; anchors_n=0 for o in obs: q=cmap[o["id"]]; anchors=q["anchors"]; anchors_n+=len(anchors) if q["status"]=="indeterminate": reasons.append("indeterminate_claim") if o["kind"]=="static" and "static" not in actions: reasons.append("unauthorized_static") if o["kind"] in ("execution","red_green") and "execute" not in actions and any(emap.get(i,{}).get("origin")=="direct" for i in q["execution_ids"]): reasons.append("unauthorized_execute") if o["kind"]=="static" and not anchors: reasons.append("missing_static_anchor") if o["kind"] in ("execution","red_green") and not q["execution_ids"]: reasons.append("missing_execution_record") for a in anchors: if not isinstance(a,dict) or "kind" not in a: _bad() if a["kind"]=="lines": _keys(a,("kind","path","start_line","end_line","text_sha256")); p=_path(a["path"]); _digest(a["text_sha256"]) if p not in files or any(isinstance(a[x],bool) or not isinstance(a[x],int) for x in ("start_line","end_line")) or a["start_line"]<=0 or a["end_line"]len(ls): reasons.append("anchor_range_invalid") elif hashlib.sha256(b"".join(ls[a["start_line"]-1:a["end_line"]])).hexdigest()!=a["text_sha256"]: reasons.append("anchor_hash_mismatch") elif a["kind"]=="test": _keys(a,("kind","path","symbol")); p=_path(a["path"]) if p not in files or not p.endswith(".py") or not _has_test(raw[p],a["symbol"]): reasons.append("test_anchor_missing") elif a["kind"]=="absence": _keys(a,("kind","path")); p=_path(a["path"]) if src["kind"]!="git_tree": reasons.append("absence_indeterminate") else: present=None if gitfd is None else _git_has_path(src,p,gitfd) if p in files or present is True: reasons.append("absence_not_proven") elif present is None: reasons.append("absence_indeterminate") else: _bad() for eid in q["execution_ids"]: if eid not in emap: reasons.append("unknown_execution_id"); continue e=emap[eid] if e["origin"]=="direct" and "execute" not in actions: reasons.append("unauthorized_execute") if e["bundle_sha256"]!=context["bundle_sha256"] and not (o["kind"]=="red_green" and e["exit_code"]!=0 and e["failure_kind"] in ("assertion","approved_behavior")): reasons.append("execution_scope_mismatch") if hashlib.sha256(_read(rootfd,e["evidence_path"])).hexdigest()!=e["evidence_sha256"]: reasons.append("evidence_hash_mismatch") if o["kind"]=="red_green": tests=[a for a in anchors if isinstance(a,dict) and a.get("kind")=="test"] if len(tests)!=1: reasons.append("red_green_test_anchor_required") else: test_id=tests[0]["path"]+"::"+tests[0]["symbol"]; red=False; green=False for eid in q["execution_ids"]: if eid not in emap: continue e=emap[eid] if e["test_id"]!=test_id or e["test_file_sha256"]!=files.get(tests[0]["path"]): reasons.append("red_green_test_scope_mismatch") if e["exit_code"]!=0 and e["failure_kind"] in ("assertion","approved_behavior"): red=True if e["exit_code"]==0 and e["failure_kind"] in ("none","approved_behavior") and e["bundle_sha256"]==context["bundle_sha256"]: green=True if not red: reasons.append("red_green_missing_red") if not green: reasons.append("red_green_missing_green") counts={"files":len(files),"obligations":len(obs),"claims":len(claims),"anchors":anchors_n,"executions":len(exs)} return {"status":"block" if reasons else "pass","reason_codes":sorted(set(reasons)),"counts":counts} except EvidenceError: raise except (OSError,TypeError,ValueError,KeyError,RecursionError): _bad() finally: if gitfd is not None: os.close(gitfd) os.close(rootfd) class _Parser(argparse.ArgumentParser): def error(self,message): _bad() def main(argv=None): p=_Parser(add_help=False,allow_abbrev=False); p.add_argument("--bundle-root",required=True); p.add_argument("--context",required=True); p.add_argument("--result",required=True); p.add_argument("--git-dir") try: a=p.parse_args(argv); out=verify_evidence(a.bundle_root,_json_file(a.context),_json_file(a.result),a.git_dir); print(json.dumps(out,sort_keys=True,separators=(",",":"))); return 0 if out["status"]=="pass" else 1 except Exception: print('{"status":"block","reason_codes":["rejected_input"],"counts":{}}'); return 2 if __name__=="__main__": raise SystemExit(main())