108 lines
5.4 KiB
Python
108 lines
5.4 KiB
Python
#!/usr/bin/env python3
|
|
"""Current-source endpoint/privilege inventory; no installed state is accessed.
|
|
|
|
Records exact lexical sites and source hashes. Dynamic arguments are explicitly
|
|
unresolved here, not misrepresented as a complete inferred authorization policy.
|
|
"""
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import re
|
|
|
|
ROOT=Path(__file__).resolve().parents[1]
|
|
PRIVILEGED={'sudo','nmcli','ip','iw','rfkill','reboot','shutdown','modprobe','mount','umount','chown','chmod','systemctl','timedatectl','udevadm'}
|
|
|
|
|
|
def mask_source(text):
|
|
"""Keep offsets/newlines while masking Rust comments and string literals."""
|
|
out=list(text);strings=[];i=0
|
|
while i<len(text):
|
|
start=i
|
|
if text.startswith('//',i):
|
|
end=text.find('\n',i);end=len(text) if end<0 else end
|
|
elif text.startswith('/*',i):
|
|
depth=1;end=i+2
|
|
while end<len(text) and depth:
|
|
if text.startswith('/*',end):depth+=1;end+=2
|
|
elif text.startswith('*/',end):depth-=1;end+=2
|
|
else:end+=1
|
|
else:
|
|
raw=re.match(r'(?:br|r)(#{0,32})"',text[i:])
|
|
if raw:
|
|
terminator='"'+raw[1];body=i+len(raw[0]);finish=text.find(terminator,body)
|
|
end=len(text) if finish<0 else finish+len(terminator)
|
|
strings.append((start,end,text[body:finish] if finish>=0 else ''))
|
|
elif text[i]=='"':
|
|
end=i+1
|
|
while end<len(text):
|
|
if text[end]=='\\':end+=2
|
|
elif text[end]=='"':end+=1;break
|
|
else:end+=1
|
|
strings.append((start,end,text[start+1:end-1]))
|
|
else:i+=1;continue
|
|
for j in range(start,end):
|
|
if out[j]!='\n':out[j]=' '
|
|
i=end
|
|
return ''.join(out),strings
|
|
|
|
|
|
def span(mask,opening):
|
|
depth=0
|
|
for i in range(opening,len(mask)):
|
|
if mask[i]=='{':depth+=1
|
|
elif mask[i]=='}':
|
|
depth-=1
|
|
if depth==0:return i+1
|
|
return len(mask)
|
|
|
|
|
|
def inventory():
|
|
sites=[];endpoints=[];hashes={}
|
|
for tree in ('core/archipelago/src','core/container/src'):
|
|
for path in sorted((ROOT/tree).rglob('*.rs')):
|
|
text=path.read_text();relative=str(path.relative_to(ROOT));mask,strings=mask_source(text)
|
|
hashes[relative]=hashlib.sha256(text.encode()).hexdigest()
|
|
functions=[];tests=[]
|
|
for m in re.finditer(r'\b(?:async\s+)?fn\s+(\w+)\s*\(',mask):
|
|
brace=mask.find('{',m.end());semi=mask.find(';',m.end())
|
|
if brace>=0 and (semi<0 or brace<semi):functions.append((m.start(),span(mask,brace),m[1]))
|
|
for m in re.finditer(r'#\s*\[\s*cfg\s*\(\s*test\s*\)\s*\]\s*(?:pub\s+)?mod\s+\w+\s*\{|\bmod\s+\w*tests\s*\{',mask):
|
|
brace=mask.find('{',m.start());tests.append((m.start(),span(mask,brace)))
|
|
def context(position):
|
|
matches=[f for f in functions if f[0]<=position<f[1]]
|
|
return max(matches,default=(0,0,'module'),key=lambda f:f[0])[2]
|
|
def testing(position):
|
|
return path.stem.endswith('_tests') or any(a<=position<b for a,b in tests)
|
|
for m in re.finditer(r'\b(host_sudo(?:_output|_with_input)?)\s*\(|\bCommand\s*::\s*new\s*\(',mask):
|
|
function=context(m.start())
|
|
if m[1] and function==m[1]:continue
|
|
literal=next((value for a,b,value in strings if m.end()<=a<m.end()+100 and not mask[m.end():a].strip()),None)
|
|
if not m[1] and literal not in PRIVILEGED:continue
|
|
line=text.count('\n',0,m.start())+1
|
|
excerpt=text[m.start():m.start()+700].split('\n\n',1)[0]
|
|
sites.append(dict(file=relative,line=line,function=function,test=testing(m.start()),mechanism=m[1] or literal,source=excerpt))
|
|
for a,b,value in strings:
|
|
for m in re.finditer(r'(?:https?://|socks5h?://)?(?:127\.0\.0\.1|localhost|\[::1\]):(\d{1,5})',value):
|
|
port=int(m[1])
|
|
if port:
|
|
endpoints.append(dict(file=relative,line=text.count('\n',0,a)+1,function=context(a),
|
|
test=testing(a),endpoint=m[0],port=port,template='proxy_pass' in value or '[Service]' in value))
|
|
return dict(version=1,repository=str(ROOT),source_sha256=hashes,privilege_sites=sites,loopback_sites=endpoints,
|
|
limits=['Lexical call inventory, not dynamic-call-graph or authorization proof.',
|
|
'Manifest-derived and configured dynamic ports require separate qualified input projection.'])
|
|
|
|
|
|
if __name__=='__main__':
|
|
parser=argparse.ArgumentParser(description=__doc__);parser.add_argument('output',type=Path);parser.add_argument('--summary',action='store_true');args=parser.parse_args()
|
|
result=inventory()
|
|
with args.output.open('x') as output:json.dump(result,output,indent=2);output.write('\n')
|
|
print(json.dumps({'production_privilege_sites':sum(not x['test'] for x in result['privilege_sites']),
|
|
'production_loopback_ports':sorted({x['port'] for x in result['loopback_sites'] if not x['test']}),
|
|
'output':str(args.output)},sort_keys=True))
|
|
if args.summary:
|
|
grouped={}
|
|
for row in result['privilege_sites']:
|
|
if not row['test']:grouped.setdefault(row['file'],set()).add(row['function']+':'+row['mechanism'])
|
|
print(json.dumps({p:sorted(v) for p,v in grouped.items()},indent=2))
|