403 lines
23 KiB
Python
403 lines
23 KiB
Python
"""Structural workflow guards; evidence claims are shape-only, not authenticated.
|
|
|
|
CLI (exactly four commands): snapshot --repo-root ROOT --paths-json INPUT
|
|
--output SNAPSHOT; compare-source --repo-root ROOT --before SNAPSHOT
|
|
--paths-json INPUT; check-tests/check-contracts --input JSON. Exit 0 is PASS,
|
|
1 detected drift/violations, 2 rejected input. APIs raise GuardError for rejected
|
|
input or return status/violations. JSON inputs are regular files bounded to 1 MiB;
|
|
unknown/missing/duplicate object keys and boolean-as-integer fields block.
|
|
ROOT is a literal canonical absolute Git root, without symlink ancestors. Paths
|
|
are a nonempty unique repository-relative string array, with no dot/traversal
|
|
segments or backslash aliases. Protected files must exist and be regular.
|
|
Snapshot output is new/no-clobber, mode 0600, outside ROOT, in an already-existing
|
|
executing-UID-owned 0700 directory. No permissions or directories are repaired.
|
|
|
|
check_tests input has exactly these required keys:
|
|
baseline_ids, current_ids: arrays of unique nonempty test-ID strings;
|
|
baseline_status, current_status: exact ID-keyed maps with values
|
|
{skip: bool, expected_failure: bool};
|
|
baseline_collection, current_collection: objects with exactly
|
|
{source_revision: lowercase 40-hex, command_id: nonempty string,
|
|
argv: nonempty array of nonempty strings, inventory_sha256: lowercase 64-hex,
|
|
evidence: nonempty reference string}.
|
|
The digest is SHA256 over UTF-8 bytes of json.dumps(
|
|
{'ids': sorted(ids), 'statuses': {id: statuses[id] for id in sorted(ids)}},
|
|
sort_keys=True, separators=(',', ':')) with default JSON escaping.
|
|
authorized_changes: array of exact objects
|
|
{old_id: baseline ID, kind: 'delete'|'rename'|'status',
|
|
replacement_ids: unique current-ID array, reason: nonempty string,
|
|
approval: nonempty reference, evidence: nonempty semantic-review reference}.
|
|
delete requires old ID absent and empty replacements; rename requires old ID
|
|
absent and at least one actual replacement; status requires old ID present
|
|
and replacements exactly [old_id]. Old IDs and replacement targets cannot
|
|
repeat between approvals. Ordinary unskipped regression additions need no
|
|
deletion approval. Every removed ID and newly introduced skip/xfail requires
|
|
a compatible approval. Do not invent replacement coverage.
|
|
changed_tests: array, unique by test_id, of exact objects
|
|
{test_id: current ID, baseline_sha256: lowercase 64-hex,
|
|
current_sha256: lowercase 64-hex, review_evidence: nonempty reference}.
|
|
findings: array, unique by id, of exact objects
|
|
{id: nonempty string, test_id: current ID,
|
|
classification: 'assertion'|'approved_behavior', red: execution,
|
|
green: execution}. Each execution has exactly
|
|
{test_id: finding's test_id, source_revision: lowercase 40-hex,
|
|
scope_sha256: lowercase 64-hex, command_id: nonempty reference,
|
|
argv: nonempty string array, exit_code: int, expected_exit_code: same int,
|
|
failure_kind: string, evidence: nonempty reference}.
|
|
RED exit is positive/nonzero, failure_kind assertion or approved_behavior;
|
|
GREEN exit is 0, failure_kind pass. Import/environment failures are not RED.
|
|
|
|
check_contracts input has exactly baseline, current, mappings, stages, precedence:
|
|
baseline: nonempty array, unique semantic_key, of exact objects
|
|
{semantic_key: nonempty string, baseline_revision: lowercase 40-hex,
|
|
path: safe repository-relative path, meaning: nonempty string};
|
|
current: array, unique semantic_key, of exact objects
|
|
{semantic_key: nonempty string, meaning: nonempty string,
|
|
observation: 'observed'|'not_applicable'|'unknown', evidence: nonempty reference}.
|
|
unknown blocks even with evidence; not_applicable needs genuine justification.
|
|
mappings: one exact object per baseline semantic_key
|
|
{semantic_key: baseline key, current_keys: unique current-key array,
|
|
status: 'preserved'|'intentionally_changed'|'missing'|'ambiguous',
|
|
authorization: string, verification: string}.
|
|
Resolved rows require nonempty current_keys/verification. intentionally_changed
|
|
additionally requires nonempty authorization; preserved may use empty
|
|
authorization. missing/ambiguous always block. All current keys are covered.
|
|
Substitutions and merges may use new current keys; several old rows can map
|
|
to one current row without dropping any old meaning or verification.
|
|
stages: nonempty array of unique nonempty strings;
|
|
precedence: array of unique [prerequisite, dependent] declared-stage pairs.
|
|
Initialization graph must be acyclic, not a graph of allowed lifecycle loops.
|
|
|
|
Snapshot JSON is generated by snapshot, not hand-declared: schema integer 1,
|
|
repo canonical root, head full SHA, sorted paths, files keyed by exact paths.
|
|
Each file object has present true, type regular, sha256 64-hex, and nonnegative
|
|
integer device/inode/mode/uid/nlink. Comparison checks HEAD, path set and file
|
|
content/identity metadata; it does not prove absence of edit-then-restore or
|
|
out-of-scope writes and is not a sandbox.
|
|
|
|
No evidence/approval reference is opened or authenticated by check_tests or
|
|
check_contracts. Main verifies actual source HEAD/blob/scope digests, approvals,
|
|
argv/environment/output/exit and intended named failure using existing immutable
|
|
verification records. Declared hashes are not proof of source identity. Meaning
|
|
preservation and omitted edits require independent review. Runnable nested JSON
|
|
examples and main-agent binding procedure are in repository-cache.md, Shared
|
|
workflow guard gates / Nested input contracts; synthetic fixtures are not evidence.
|
|
"""
|
|
import argparse, hashlib, json, os, re, stat, subprocess
|
|
from functools import wraps
|
|
from pathlib import Path
|
|
|
|
class GuardError(ValueError): pass
|
|
|
|
_HEX40 = re.compile(r'^[0-9a-f]{40}$')
|
|
_HEX64 = re.compile(r'^[0-9a-f]{64}$')
|
|
|
|
def _fail(message='invalid workflow guard input'):
|
|
raise GuardError(message)
|
|
|
|
def _keys(obj, keys):
|
|
if not isinstance(obj, dict) or set(obj) != set(keys): _fail('invalid object fields')
|
|
|
|
def _str(v): return isinstance(v, str) and bool(v.strip())
|
|
def _hex(v, pattern): return isinstance(v, str) and pattern.fullmatch(v) is not None
|
|
|
|
def _guard_errors(function):
|
|
@wraps(function)
|
|
def checked(*args, **kwargs):
|
|
try:
|
|
return function(*args, **kwargs)
|
|
except GuardError:
|
|
raise
|
|
except (OSError, ValueError, TypeError, KeyError, RecursionError):
|
|
_fail('workflow guard rejected input')
|
|
return checked
|
|
|
|
def _directory(raw):
|
|
if not isinstance(raw, str) or not os.path.isabs(raw) or raw != os.path.normpath(raw):
|
|
_fail('noncanonical directory path')
|
|
if raw != '/' and any(part in ('', '.', '..') for part in raw.split('/')[1:]):
|
|
_fail('noncanonical directory path')
|
|
fd = os.open('/', os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
|
|
try:
|
|
for part in raw.split('/')[1:]:
|
|
if not part:
|
|
continue
|
|
nxt = os.open(part, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=fd)
|
|
os.close(fd)
|
|
fd = nxt
|
|
return fd
|
|
except BaseException:
|
|
os.close(fd)
|
|
raise
|
|
|
|
def _json(path):
|
|
def pairs(items):
|
|
out = {}
|
|
for k,v in items:
|
|
if k in out: _fail('duplicate JSON key')
|
|
out[k]=v
|
|
return out
|
|
try:
|
|
fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK)
|
|
try:
|
|
if not stat.S_ISREG(os.fstat(fd).st_mode):
|
|
_fail('JSON input must be regular')
|
|
with os.fdopen(fd, 'rb') as f:
|
|
fd = None
|
|
raw = f.read(1048577)
|
|
finally:
|
|
if fd is not None:
|
|
os.close(fd)
|
|
if len(raw)>1048576: _fail('JSON input too large')
|
|
return json.loads(raw.decode('utf-8'), object_pairs_hook=pairs,
|
|
parse_constant=lambda _: _fail('invalid JSON number'))
|
|
except GuardError: raise
|
|
except (OSError, UnicodeError, ValueError, TypeError, RecursionError): _fail('invalid JSON input')
|
|
|
|
def _paths(paths):
|
|
if not isinstance(paths,list) or not paths or any(not isinstance(p,str) or not p for p in paths): _fail('invalid paths')
|
|
for p in paths:
|
|
if p.startswith('/') or '\\' in p or any(x in ('','.','..') for x in p.split('/')): _fail('unsafe path')
|
|
if len(set(paths)) != len(paths): _fail('duplicate path')
|
|
return sorted(paths)
|
|
|
|
def _root(root):
|
|
raw=os.fspath(root)
|
|
if not isinstance(raw,str) or not os.path.isabs(raw) or raw.endswith('/') or '//' in raw or '/./' in raw or '/../' in raw: _fail('invalid repository root')
|
|
fd=None
|
|
try:
|
|
fd=_directory(raw)
|
|
p=subprocess.run(['git','-C',raw,'rev-parse','--show-toplevel','--is-bare-repository'],check=True,stdout=subprocess.PIPE,stderr=subprocess.DEVNULL,text=True)
|
|
lines=p.stdout.splitlines()
|
|
if len(lines)!=2 or lines[1]!='false' or os.path.realpath(raw)!=lines[0]: _fail('root is not repository top-level')
|
|
return os.path.realpath(raw)
|
|
except (OSError, subprocess.SubprocessError): _fail('cannot identify repository root')
|
|
finally:
|
|
if fd is not None: os.close(fd)
|
|
|
|
def _head(root):
|
|
try:
|
|
h=subprocess.run(['git','-C',root,'rev-parse','--verify','HEAD'],check=True,stdout=subprocess.PIPE,stderr=subprocess.DEVNULL,text=True).stdout.strip()
|
|
if not _hex(h,_HEX40): _fail('invalid HEAD')
|
|
return h
|
|
except (OSError,subprocess.SubprocessError): _fail('cannot identify HEAD')
|
|
|
|
def _open_file(root,rel):
|
|
fd=_directory(root)
|
|
try:
|
|
parts=rel.split('/')
|
|
for part in parts[:-1]:
|
|
nxt=os.open(part,os.O_RDONLY|os.O_DIRECTORY|os.O_NOFOLLOW,dir_fd=fd); os.close(fd); fd=nxt
|
|
named = os.stat(parts[-1], dir_fd=fd, follow_symlinks=False)
|
|
if not stat.S_ISREG(named.st_mode):
|
|
_fail('protected path is not a regular file')
|
|
f = os.open(parts[-1], os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=fd)
|
|
try:
|
|
opened = os.fstat(f)
|
|
if not stat.S_ISREG(opened.st_mode) or (named.st_dev, named.st_ino) != (opened.st_dev, opened.st_ino):
|
|
_fail('protected file changed')
|
|
return f, opened
|
|
except BaseException:
|
|
os.close(f)
|
|
raise
|
|
finally: os.close(fd)
|
|
|
|
def _record(root, rel):
|
|
fd,st=_open_file(root,rel)
|
|
try:
|
|
h=hashlib.sha256()
|
|
while True:
|
|
b=os.read(fd,131072)
|
|
if not b: break
|
|
h.update(b)
|
|
end=os.fstat(fd)
|
|
if (st.st_dev,st.st_ino,st.st_size,st.st_mtime_ns)!=(end.st_dev,end.st_ino,end.st_size,end.st_mtime_ns): _fail('protected file changed during read')
|
|
return {'present':True,'type':'regular','sha256':h.hexdigest(),'device':st.st_dev,'inode':st.st_ino,'mode':stat.S_IMODE(st.st_mode),'uid':st.st_uid,'nlink':st.st_nlink}
|
|
finally: os.close(fd)
|
|
|
|
def _out_parent(output,root):
|
|
raw=os.path.abspath(os.fspath(output)); parent,name=os.path.split(raw)
|
|
if not name or os.path.commonpath([root,raw])==root: _fail('output must be outside repository')
|
|
cur=os.open('/' if raw.startswith('/') else '.',os.O_RDONLY|os.O_DIRECTORY)
|
|
try:
|
|
for part in [x for x in parent.split('/') if x]:
|
|
n=os.open(part,os.O_RDONLY|os.O_DIRECTORY|os.O_NOFOLLOW,dir_fd=cur); os.close(cur); cur=n
|
|
st=os.fstat(cur)
|
|
if st.st_uid!=os.getuid() or stat.S_IMODE(st.st_mode)!=0o700: _fail('output parent must be owned 0700 directory')
|
|
return cur,name
|
|
except (OSError, GuardError):
|
|
os.close(cur); _fail('unsafe output parent')
|
|
|
|
@_guard_errors
|
|
def snapshot(repo_root,paths,output):
|
|
root=_root(repo_root); paths=_paths(paths); head=_head(root)
|
|
files={p:_record(root,p) for p in paths}
|
|
if _head(root)!=head: _fail('HEAD changed during capture')
|
|
parent,name=_out_parent(output,root)
|
|
fd = None
|
|
try:
|
|
fd=os.open(name,os.O_WRONLY|os.O_CREAT|os.O_EXCL|os.O_NOFOLLOW,0o600,dir_fd=parent)
|
|
os.fchmod(fd,0o600); st=os.fstat(fd); named=os.stat(name,dir_fd=parent,follow_symlinks=False)
|
|
if (st.st_dev,st.st_ino)!=(named.st_dev,named.st_ino): _fail('output changed')
|
|
data={'schema':1,'repo':root,'head':head,'paths':paths,'files':files}
|
|
with os.fdopen(fd,'w',encoding='utf8') as f:
|
|
fd = None
|
|
json.dump(data,f,sort_keys=True); f.write('\n')
|
|
except (OSError,ValueError,TypeError): _fail('cannot safely create snapshot')
|
|
finally:
|
|
if fd is not None: os.close(fd)
|
|
os.close(parent)
|
|
return {'status':'pass','head':head,'paths':paths}
|
|
|
|
@_guard_errors
|
|
def compare_source(repo_root,before,paths):
|
|
root=_root(repo_root); paths=_paths(paths); b=_json(before)
|
|
_keys(b,('schema','repo','head','paths','files'))
|
|
if type(b['schema']) is not int or b['schema']!=1 or b['repo']!=root or not _hex(b['head'],_HEX40) or b['paths']!=paths or not isinstance(b['files'],dict) or set(b['files'])!=set(paths): _fail('malformed source baseline')
|
|
for x in b['files'].values():
|
|
_keys(x,('present','type','sha256','device','inode','mode','uid','nlink'))
|
|
if x['present'] is not True or x['type']!='regular' or not _hex(x['sha256'],_HEX64) or any(type(x[k]) is not int or x[k]<0 for k in ('device','inode','mode','uid','nlink')): _fail('malformed file baseline')
|
|
violations=[]
|
|
if _head(root)!=b['head']: violations.append('HEAD changed')
|
|
for p in paths:
|
|
try:
|
|
r=_record(root,p)
|
|
if r!=b['files'][p]: violations.append('source changed: '+p)
|
|
except (GuardError, OSError): violations.append('source missing or changed: '+p)
|
|
return {'status':'block' if violations else 'pass','violations':violations}
|
|
|
|
def _ids(d,k):
|
|
x=d.get(k)
|
|
if not isinstance(x,list) or any(not _str(i) for i in x) or len(set(x))!=len(x): _fail('invalid IDs')
|
|
return x,set(x)
|
|
|
|
def _status(s):
|
|
_keys(s,('skip','expected_failure'))
|
|
if type(s['skip']) is not bool or type(s['expected_failure']) is not bool: _fail('invalid status')
|
|
|
|
def _collection(c,ids,status):
|
|
_keys(c,('source_revision','command_id','argv','inventory_sha256','evidence'))
|
|
if not _hex(c['source_revision'],_HEX40) or not _str(c['command_id']) or not isinstance(c['argv'],list) or not c['argv'] or any(not _str(a) for a in c['argv']) or not _str(c['evidence']): _fail('invalid collection provenance')
|
|
digest=hashlib.sha256(json.dumps({'ids':sorted(ids),'statuses':{i:status[i] for i in sorted(ids)}},sort_keys=True,separators=(',',':')).encode()).hexdigest()
|
|
if c['inventory_sha256']!=digest: _fail('collection digest mismatch')
|
|
|
|
@_guard_errors
|
|
def check_tests(data):
|
|
_keys(data,('baseline_ids','current_ids','baseline_status','current_status','baseline_collection','current_collection','authorized_changes','changed_tests','findings'))
|
|
old,olds=_ids(data,'baseline_ids'); new,news=_ids(data,'current_ids')
|
|
violations=[]
|
|
for prefix,ids in (('baseline',olds),('current',news)):
|
|
statuses=data[prefix+'_status']
|
|
if not isinstance(statuses,dict) or set(statuses)!=ids: _fail('status inventory mismatch')
|
|
for s in statuses.values(): _status(s)
|
|
_collection(data[prefix+'_collection'],ids,statuses)
|
|
auth=data['authorized_changes']
|
|
if not isinstance(auth,list): _fail('invalid authorizations')
|
|
approvals={}; targets=set()
|
|
for a in auth:
|
|
_keys(a,('old_id','kind','replacement_ids','reason','approval','evidence'))
|
|
if not _str(a['old_id']) or a['old_id'] not in olds or a['old_id'] in approvals or not isinstance(a['replacement_ids'],list) or any(not _str(x) or x not in news for x in a['replacement_ids']) or len(set(a['replacement_ids']))!=len(a['replacement_ids']) or any(not _str(a[k]) for k in ('reason','approval','evidence')): _fail('invalid authorization')
|
|
kind, old_id, replacements = a['kind'], a['old_id'], a['replacement_ids']
|
|
if kind == 'delete':
|
|
if old_id in news or replacements: _fail('invalid deletion authorization')
|
|
elif kind == 'rename':
|
|
if old_id in news or not replacements: _fail('invalid rename authorization')
|
|
elif kind == 'status':
|
|
if old_id not in news or replacements != [old_id]: _fail('invalid status authorization')
|
|
else: _fail('invalid authorization kind')
|
|
if targets.intersection(a['replacement_ids']): _fail('reused replacement')
|
|
targets.update(a['replacement_ids']); approvals[a['old_id']]=a
|
|
for i in olds-news:
|
|
if i not in approvals: violations.append('removed test lacks authorization')
|
|
for i in news:
|
|
if data['current_status'][i]['skip'] and (i not in olds or not data['baseline_status'][i]['skip']) and not any(i in a['replacement_ids'] or a['old_id']==i for a in auth): violations.append('new skip lacks authorization')
|
|
if data['current_status'][i]['expected_failure'] and (i not in olds or not data['baseline_status'][i]['expected_failure']) and not any(i in a['replacement_ids'] or a['old_id']==i for a in auth): violations.append('new expected failure lacks authorization')
|
|
changed=data['changed_tests']
|
|
if not isinstance(changed,list): _fail('invalid changed tests')
|
|
seen=set()
|
|
for c in changed:
|
|
_keys(c,('test_id','baseline_sha256','current_sha256','review_evidence'))
|
|
i=c['test_id']
|
|
if i not in news or i in seen or not _hex(c['baseline_sha256'],_HEX64) or not _hex(c['current_sha256'],_HEX64) or not _str(c['review_evidence']): _fail('invalid changed test review')
|
|
seen.add(i)
|
|
findings=data['findings']
|
|
if not isinstance(findings,list): _fail('invalid findings')
|
|
fids=set()
|
|
for f in findings:
|
|
_keys(f,('id','test_id','classification','red','green'))
|
|
if not _str(f['id']) or f['id'] in fids or f['test_id'] not in news or f['classification'] not in ('assertion','approved_behavior'): _fail('invalid finding')
|
|
fids.add(f['id'])
|
|
for stage in ('red','green'):
|
|
r=f[stage]; _keys(r,('test_id','source_revision','scope_sha256','command_id','argv','exit_code','expected_exit_code','failure_kind','evidence'))
|
|
if r['test_id']!=f['test_id'] or not _hex(r['source_revision'],_HEX40) or not _hex(r['scope_sha256'],_HEX64) or not _str(r['command_id']) or not isinstance(r['argv'],list) or not r['argv'] or any(not _str(a) for a in r['argv']) or type(r['exit_code']) is not int or r['expected_exit_code']!=r['exit_code'] or type(r['expected_exit_code']) is not int or not _str(r['evidence']): _fail('invalid finding execution')
|
|
if stage=='red' and (r['exit_code']<=0 or r['failure_kind'] not in ('assertion','approved_behavior')): _fail('invalid RED evidence')
|
|
if stage=='green' and (r['exit_code']!=0 or r['failure_kind']!='pass'): _fail('invalid GREEN evidence')
|
|
return {'status':'block' if violations else 'pass','violations':violations}
|
|
|
|
@_guard_errors
|
|
def check_contracts(data):
|
|
_keys(data,('baseline','current','mappings','stages','precedence'))
|
|
b,c,m=data['baseline'],data['current'],data['mappings']
|
|
if not all(isinstance(x,list) for x in (b,c,m)) or not b: _fail('invalid or empty contracts')
|
|
def unique_rows(rows,base):
|
|
out={}
|
|
for r in rows:
|
|
fields=('semantic_key','baseline_revision','path','meaning') if base else ('semantic_key','meaning','observation','evidence')
|
|
_keys(r,fields); k=r['semantic_key']
|
|
if not _str(k) or k in out or not _str(r['meaning']): _fail('invalid contract row')
|
|
if base:
|
|
if not _hex(r['baseline_revision'],_HEX40) or not isinstance(r['path'],str) or not r['path'] or r['path'].startswith('/') or any(x in ('','.','..') for x in r['path'].split('/')): _fail('invalid baseline contract')
|
|
else:
|
|
if r['observation'] not in ('observed','unknown','not_applicable') or not isinstance(r['evidence'],str) or not r['evidence'].strip() or r['observation']=='unknown': _fail('unverified current contract')
|
|
out[k]=r
|
|
return out
|
|
old=unique_rows(b,True); new=unique_rows(c,False)
|
|
violations=[]; maps={}
|
|
for r in m:
|
|
_keys(r,('semantic_key','current_keys','status','authorization','verification'))
|
|
k=r['semantic_key']
|
|
if k not in old or k in maps or not isinstance(r['current_keys'],list) or any(not isinstance(x,str) for x in r['current_keys']) or len(set(r['current_keys']))!=len(r['current_keys']) or any(x not in new for x in r['current_keys']) or not isinstance(r['authorization'],str) or not isinstance(r['verification'],str): _fail('invalid mapping')
|
|
maps[k]=r
|
|
if set(maps)!=set(old): _fail('incomplete mappings')
|
|
claimed=set()
|
|
for k,r in maps.items():
|
|
status=r['status']; keys=r['current_keys']
|
|
if status not in ('preserved','intentionally_changed','missing','ambiguous'): _fail('invalid mapping status')
|
|
if status in ('missing','ambiguous'): violations.append('unresolved contract: '+k)
|
|
elif not keys or not r['verification'].strip(): violations.append('contract mapping lacks verification: '+k)
|
|
if status=='intentionally_changed' and not r['authorization'].strip(): violations.append('contract change lacks authorization: '+k)
|
|
claimed.update(keys)
|
|
if claimed != set(new): _fail('unmapped current contract')
|
|
stages=data['stages']; edges=data['precedence']
|
|
if not isinstance(stages,list) or not stages or any(not _str(x) for x in stages) or len(set(stages))!=len(stages): _fail('invalid stages')
|
|
if not isinstance(edges,list): _fail('invalid precedence')
|
|
graph={x:[] for x in stages}; indegree={x:0 for x in stages}; seen=set()
|
|
for e in edges:
|
|
if not isinstance(e,list) or len(e)!=2 or any(not isinstance(x,str) or x not in graph for x in e) or tuple(e) in seen: _fail('invalid precedence edge')
|
|
seen.add(tuple(e)); graph[e[0]].append(e[1]); indegree[e[1]]+=1
|
|
queue=[x for x in stages if indegree[x]==0]; n=0
|
|
while queue:
|
|
x=queue.pop(); n+=1
|
|
for y in graph[x]:
|
|
indegree[y]-=1
|
|
if indegree[y]==0: queue.append(y)
|
|
if n!=len(stages): violations.append('precedence cycle')
|
|
return {'status':'block' if violations else 'pass','violations':violations}
|
|
|
|
def main(argv=None):
|
|
p=argparse.ArgumentParser(); s=p.add_subparsers(dest='cmd',required=True)
|
|
x=s.add_parser('snapshot'); x.add_argument('--repo-root',required=True); x.add_argument('--paths-json',required=True); x.add_argument('--output',required=True)
|
|
x=s.add_parser('compare-source'); x.add_argument('--repo-root',required=True); x.add_argument('--before',required=True); x.add_argument('--paths-json',required=True)
|
|
for name in ('check-tests','check-contracts'):
|
|
x=s.add_parser(name); x.add_argument('--input',required=True)
|
|
a=p.parse_args(argv)
|
|
try:
|
|
if a.cmd=='snapshot': result=snapshot(a.repo_root,_json(a.paths_json),a.output)
|
|
elif a.cmd=='compare-source': result=compare_source(a.repo_root,a.before,_json(a.paths_json))
|
|
elif a.cmd=='check-tests': result=check_tests(_json(a.input))
|
|
else: result=check_contracts(_json(a.input))
|
|
print(json.dumps(result,sort_keys=True)); return 0 if result['status']=='pass' else 1
|
|
except (GuardError,OSError,ValueError,TypeError,RecursionError):
|
|
print(json.dumps({'status':'block','violations':['workflow guard rejected input']},sort_keys=True)); return 2
|
|
if __name__=='__main__': raise SystemExit(main())
|