tegwick
52168c3bd4
Some checks failed
Use-case regression gate / usecase-regressions (push) Has been cancelled
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a18e2d-f513-7742-b8e7-12af0b744ff3
190 lines
8.6 KiB
Python
190 lines
8.6 KiB
Python
#!/usr/bin/env python3
|
|
"""Prepare an offline private access inventory; never infer target identities or grants."""
|
|
|
|
import argparse
|
|
import hashlib
|
|
import html
|
|
import json
|
|
import os
|
|
import re
|
|
import stat
|
|
import sys
|
|
from collections import Counter
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
from coulomb_social.apps.spaces.bubble_transfer import normalize_ref # noqa: E402
|
|
|
|
FIELDS = {
|
|
'spaces': {'Created By': False, 'crew_list_user': True,
|
|
'members_list_user': True, 'guests_list_user': True},
|
|
'articles': {'Created By': False, '_creator_user': False,
|
|
'_owner_user': False, 'moders_list_user': True},
|
|
}
|
|
OPTIONS = {'spaces': ['isprivate_boolean', 'addnewusersasmember_boolean',
|
|
'addnewusersasguest_boolean'],
|
|
'articles': ['access_option_visibility']}
|
|
|
|
|
|
def sha(raw):
|
|
return hashlib.sha256(raw).hexdigest()
|
|
|
|
|
|
def private_read(path):
|
|
if any(p.is_symlink() for p in (path, *path.parents)):
|
|
raise ValueError('source symlink')
|
|
info = path.stat()
|
|
if (not stat.S_ISREG(info.st_mode) or info.st_uid != os.getuid()
|
|
or info.st_mode & 0o077 or info.st_nlink != 1 or info.st_size > 16 * 1024**2):
|
|
raise ValueError('private bounded regular source required')
|
|
return path.read_bytes()
|
|
|
|
|
|
def reference(value):
|
|
try:
|
|
result = normalize_ref(value)
|
|
except (ValueError, TypeError):
|
|
return None
|
|
return result if re.fullmatch(r'[0-9]+x[0-9]+', result) else None
|
|
|
|
|
|
def field_state(row, field, many):
|
|
if field not in row:
|
|
return {'state': 'missing', 'principals': []}
|
|
raw = row[field]
|
|
if raw is None:
|
|
return {'state': 'null', 'raw': raw, 'principals': []}
|
|
if many and not isinstance(raw, list):
|
|
return {'state': 'invalid_container', 'raw': raw, 'principals': []}
|
|
if not many and not isinstance(raw, str):
|
|
return {'state': 'invalid_container', 'raw': raw, 'principals': []}
|
|
values = raw if many else [raw]
|
|
refs = [reference(v) for v in values]
|
|
return {'state': ('empty' if not values else 'unresolved_reference'
|
|
if None in refs else 'observed'),
|
|
'raw': raw, 'principals': sorted(set(r for r in refs if r is not None)),
|
|
'unresolved_count': sum(r is None for r in refs)}
|
|
|
|
|
|
def audit(inputs):
|
|
"""Source values remain private; the separate summary contains aggregate counts only."""
|
|
records = {}
|
|
principals = set()
|
|
summary = {}
|
|
for kind in FIELDS:
|
|
source = inputs[kind]
|
|
if not isinstance(source, dict) or any(not isinstance(v, dict) for v in source.values()):
|
|
raise ValueError('record dictionaries required')
|
|
records[kind] = []
|
|
counters = {f: Counter() for f in FIELDS[kind]}
|
|
for key, row in sorted(source.items()):
|
|
fields = {f: field_state(row, f, many) for f, many in FIELDS[kind].items()}
|
|
for f, value in fields.items():
|
|
counters[f][value['state']] += 1
|
|
principals.update(value['principals'])
|
|
records[kind].append({'source_id': key, 'fields': fields,
|
|
'options': {f: {'present': f in row, 'value': row.get(f)}
|
|
for f in OPTIONS[kind]}})
|
|
summary[kind] = {
|
|
'records': len(source), 'field_states': counters,
|
|
'option_fields': {f: {
|
|
'present': sum(f in row for row in source.values()),
|
|
'missing': sum(f not in row for row in source.values()),
|
|
'distinct_observed_values': len({json.dumps(row[f], sort_keys=True)
|
|
for row in source.values() if f in row}),
|
|
} for f in OPTIONS[kind]},
|
|
}
|
|
retained = inputs['retained_spaces']
|
|
if not isinstance(retained, dict):
|
|
raise ValueError('retained inventory required')
|
|
known = set(retained) | set(inputs['spaces'])
|
|
summary.update({'known_spaces': len(known), 'fresh_space_records': len(inputs['spaces']),
|
|
'inventory_only_spaces': len(set(retained) - set(inputs['spaces'])),
|
|
'distinct_source_principals': len(principals),
|
|
'target_identity_mappings_established': 0,
|
|
'target_grants_proposed': 0, 'production_changed': False,
|
|
'source_visibility_semantics_verified': False})
|
|
return {'schema': 'coulomb.bubble-access-inventory.v1', 'summary': summary,
|
|
'source_principals': sorted(principals), 'records': records,
|
|
'inventory_only_space_ids': sorted(set(retained) - set(inputs['spaces'])),
|
|
'limitations': [
|
|
'Missing fields are unknown, not empty grants.',
|
|
'Source roles and visibility are observed values, not target permissions.',
|
|
'Existing one-space owner admission does not map every source identity.',
|
|
'This capture does not prove complete or current provider membership.',
|
|
'No target identity, role mapping or import manifest is created.']}
|
|
|
|
|
|
def review_html(report):
|
|
rows = []
|
|
for kind, records in report['records'].items():
|
|
for row in records:
|
|
data = json.dumps(row, ensure_ascii=False, indent=2)
|
|
rows.append('<details><summary>' + html.escape(kind + ': ' + row['source_id'])
|
|
+ '</summary><pre>' + html.escape(data) + '</pre></details>')
|
|
return ('<!doctype html><html><head><meta charset="utf-8">'
|
|
'<meta http-equiv="Content-Security-Policy" content="default-src \'none\'; '
|
|
'base-uri \'none\'; form-action \'none\'">'
|
|
'<title>Private source access inventory</title></head><body>'
|
|
'<h1>Private source access inventory</h1><p>Observed source metadata only. '
|
|
'No target identities or grants have been inferred.</p><pre>'
|
|
+ html.escape(json.dumps(report['summary'], indent=2)) + '</pre><ul>'
|
|
+ ''.join('<li>' + html.escape(x) + '</li>' for x in report['limitations'])
|
|
+ '</ul>' + ''.join(rows) + '</body></html>').encode()
|
|
|
|
|
|
def prepare(source, out):
|
|
raw = {kind: private_read(source / 'source-capture' / (kind + '.json'))
|
|
for kind in [*FIELDS, 'retained_spaces']}
|
|
receipt_raw = private_read(source / 'reconciliation.json')
|
|
receipt = json.loads(receipt_raw)
|
|
if (receipt['schema'] != 'coulomb.bubble-review-bundle.v1'
|
|
or receipt['status'] != 'completed_offline_preparation'):
|
|
raise ValueError('completed review bundle required')
|
|
hashes = {k: sha(v) for k, v in raw.items()}
|
|
if any(receipt['source_sha256'].get(k) != v for k, v in hashes.items()):
|
|
raise ValueError('source digest mismatch')
|
|
report = audit({k: json.loads(v) for k, v in raw.items()})
|
|
report['source_sha256'] = hashes
|
|
report['source_report_sha256'] = sha(receipt_raw)
|
|
if out.exists() or any(p.is_symlink() for p in (out, *out.parents)):
|
|
raise ValueError('new private output required')
|
|
parent = out.parent.stat()
|
|
if (parent.st_uid != os.getuid() or parent.st_mode & 0o077
|
|
or out.resolve().is_relative_to(Path(__file__).resolve().parents[1])
|
|
or out.resolve().is_relative_to(source.resolve())):
|
|
raise ValueError('private output outside repository and source required')
|
|
for kind, data in raw.items():
|
|
if private_read(source / 'source-capture' / (kind + '.json')) != data:
|
|
raise ValueError('source changed during audit')
|
|
if private_read(source / 'reconciliation.json') != receipt_raw:
|
|
raise ValueError('source report changed during audit')
|
|
out.mkdir(mode=0o700)
|
|
outputs = {'access-inventory.json': json.dumps(report, indent=2).encode(),
|
|
'index.html': review_html(report)}
|
|
for name, data in outputs.items():
|
|
with os.fdopen(os.open(out / name, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600), 'wb') as f:
|
|
f.write(data)
|
|
f.flush()
|
|
os.fsync(f.fileno())
|
|
return {'schema': 'coulomb.bubble-access-audit-result.v1', **report['summary'],
|
|
'source_sha256': hashes, 'source_report_sha256': sha(receipt_raw),
|
|
'output_sha256': {k: sha(v) for k, v in outputs.items()}}
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('source', type=Path)
|
|
parser.add_argument('output', type=Path)
|
|
args = parser.parse_args()
|
|
try:
|
|
result = prepare(args.source, args.output)
|
|
except Exception:
|
|
print('Access audit refused; no identity or grant inferred.', file=sys.stderr)
|
|
return 1
|
|
print(json.dumps(result))
|
|
return 0
|
|
|
|
|
|
if __name__ == '__main__':
|
|
raise SystemExit(main())
|