prj-forgejo-org-refactor/tools/inventory.py

#!/usr/bin/env python3
"""Read-only org-reference inventory. Emits reference tokens, never source lines."""
import argparse
import concurrent.futures
import hashlib
import io
import zipfile
import json
import re
import subprocess
import urllib.error
import urllib.request
from pathlib import Path

BASE = 'https://forgejo.coulomb.social'
# Capture only the owner and first path component; never URL credentials/query.
REF = re.compile(r'(?<![A-Za-z0-9_-])coulomb/([A-Za-z0-9_.-]+)')
SENSITIVE = re.compile(r'(^|/)(\.env($|\.)|id_rsa|id_ed25519)|\.(pem|key|p12|pfx|age)$|sops\.|\.sops\.', re.I)
BINARY = {'.png','.jpg','.jpeg','.gif','.pdf','.zip','.gz','.woff','.woff2','.ico','.mp4','.sqlite','.db'}
SELF = Path(__file__).resolve().parents[1]


def run(repo, *args):
    r = subprocess.run(['git', '-C', str(repo), *args], capture_output=True)
    if r.returncode:
        raise RuntimeError('git ' + args[0] + ' failed')
    return r.stdout


def get(url):
    try:
        with urllib.request.urlopen(url, timeout=30) as r:
            return json.load(r)
    except urllib.error.HTTPError as e:
        raise RuntimeError('HTTP ' + str(e.code)) from None


def pages(path):
    rows = []
    seen = set()
    for page in range(1, 1001):
        batch = get(BASE + path + f'?limit=50&page={page}')
        if not isinstance(batch, list):
            raise RuntimeError('expected paginated list')
        if not batch:
            return rows
        digest = hashlib.sha256(json.dumps(batch, sort_keys=True).encode()).hexdigest()
        if digest in seen:
            raise RuntimeError('pagination repeated a page')
        seen.add(digest)
        rows.extend(batch)
    raise RuntimeError('pagination limit exceeded')


def categories(path):
    p = path.lower()
    kinds = []
    for flag, name in [
        ('argocd' in p, 'gitops'),
        (p.endswith('chart.yaml'), 'chart-metadata'),
        ('chart' in p or 'values' in p, 'chart-values'),
        (p.endswith('.gitmodules'), 'submodule'),
        (any(x in p for x in ['.gitea/','.forgejo/','.github/','workflow','jenkins','ci/']), 'ci'),
        (any(x in p for x in ['retention','prune','live-image','release','rollback']), 'release-retention'),
        (any(x in p for x in ['credential','policy','routing','access']), 'access-policy'),
        (any(x in p for x in ['lock','requirements','pyproject','package.json','npmrc']), 'package-consumer'),
        (any(x in p.split('/') for x in ['history','archived','evidence','workplans']), 'historical-or-planning'),
    ]:
        if flag:
            kinds.append(name)
    return kinds or ['other']


def scan(path):
    errors, exclusions, refs = [], [], []
    start = run(path, 'rev-parse', 'HEAD').decode().strip()
    remotes = []
    raw = run(path, 'config', '--get-regexp', r'^remote\..*\.url$') if run(path,'remote').strip() else b''
    for line in raw.decode(errors='replace').splitlines():
        key, value = line.split(' ', 1)
        match = REF.search(value)
        remotes.append({'remote':key.split('.')[1], 'owner_path':match.group(0).removesuffix('.git') if match else None,
                        'transport':'https' if value.startswith('https:') else 'ssh-or-other'})
    files = sorted(set(run(path, 'ls-files', '-z', '--cached', '--others', '--exclude-standard').split(b'\0')) - {b''})
    h = hashlib.sha256()
    for encoded in files:
        name = encoded.decode(errors='replace')
        file = path / name
        if file.is_dir():
            exclusions.append({'path':name,'reason':'gitlink-or-directory-not-scanned'}); continue
        if file.is_symlink():
            exclusions.append({'path':name,'reason':'symlink-not-followed'}); continue
        if SENSITIVE.search(name):
            exclusions.append({'path':name,'reason':'sensitive-or-encrypted'}); continue
        if file.suffix.lower() in BINARY:
            exclusions.append({'path':name,'reason':'binary-extension'}); continue
        try:
            if file.stat().st_size > 4_000_000:
                exclusions.append({'path':name,'reason':'over-4MB'}); continue
            data = file.read_bytes()
        except OSError:
            errors.append({'path':name,'reason':'unreadable-or-missing'}); continue
        h.update(encoded + b'\0' + hashlib.sha256(data).digest())
        if b'\0' in data:
            exclusions.append({'path':name,'reason':'binary-content'}); continue
        for number, line in enumerate(data.decode(errors='replace').splitlines(), 1):
            for match in REF.finditer(line):
                refs.append({'path':name,'line':number,'reference':match.group(0), 'categories':categories(name)})
    end = run(path, 'rev-parse', 'HEAD').decode().strip()
    if start != end:
        errors.append({'reason':'HEAD-changed-during-scan'})
    return {'name':path.name, 'path':str(path), 'head':start,
            'content_sha256':h.hexdigest(), 'remotes':remotes, 'files_considered':len(files),
            'references':refs,'exclusions':exclusions,'errors':errors}



def scan_archive(repo):
    """Read a public default-branch archive in memory, never extract to disk."""
    name=repo['name']
    branch=repo['default_branch']
    from urllib.parse import quote
    branch_info=get(BASE+'/api/v1/repos/coulomb/'+name+'/branches/'+quote(branch,safe=''))
    commit=branch_info['commit']['id']
    url=BASE+'/api/v1/repos/coulomb/'+name+'/archive/'+commit+'.zip'
    with urllib.request.urlopen(url,timeout=30) as response:
        raw=response.read(50_000_001)
    if len(raw)>50_000_000:
        raise RuntimeError('archive exceeds 50MB')
    refs=[];exclusions=[]
    with zipfile.ZipFile(io.BytesIO(raw)) as archive:
        for item in sorted(archive.infolist(),key=lambda i:i.filename):
            if item.is_dir(): continue
            path=item.filename.split('/',1)[-1]
            if SENSITIVE.search(path) or Path(path).suffix.lower() in BINARY or item.file_size>4_000_000:
                exclusions.append({'path':path,'reason':'sensitive-binary-or-large'});continue
            data=archive.read(item)
            if b'\0' in data:continue
            for number,line in enumerate(data.decode(errors='replace').splitlines(),1):
                for match in REF.finditer(line):
                    refs.append({'path':path,'line':number,'reference':match.group(0),'categories':categories(path)})
    return {'name':name,'commit':commit,'references':refs,'exclusions':exclusions}


def collect(root):
    gaps = [
        'Anonymous forge enumeration cannot attest private repository/package completeness.',
        'Team membership, deploy keys, webhooks, runners, protected branches and server-side access policies require an authorized owner export.',
        'Local working trees and public remote-only default-branch archives are scanned; other branches and inaccessible repos are not content-scanned.',
        'Hidden sibling directories, ignored files, submodule contents outside discovered repos, binary/large/encrypted files and symlinks are not scanned; exclusions are recorded.',
        'Runtime image/source inventory is collected separately; source scan cannot attest all runtime or external consumers.',
        'Dynamic/computed org or repository names require owner review; token scan finds literal coulomb/ references only.',
    ]
    hub = get('http://127.0.0.1:8000/repos/')
    repos = pages('/api/v1/orgs/coulomb/repos')
    try:
        packages = pages('/api/v1/packages/coulomb')
    except (RuntimeError, OSError):
        packages = []; gaps.append('Package API unavailable.')
    paths = {p.resolve() for p in root.iterdir() if p.is_dir() and not p.name.startswith('.') and (p/'.git').exists()}
    paths.update(Path(r['local_path']).resolve() for r in hub if r.get('local_path') and (Path(r['local_path'])/'.git').exists())
    paths.discard(SELF)
    with concurrent.futures.ThreadPoolExecutor(max_workers=6) as pool:
        scans = list(pool.map(scan, sorted(paths)))
    server = [{k:r.get(k) for k in ['id','name','full_name','private','archived','default_branch','description']} for r in repos]
    registered = [{k:r.get(k) for k in ['id','slug','local_path','status','description']} for r in hub]
    local_names = {m['owner_path'] for s in scans for m in s['remotes'] if m['owner_path']}
    public_names = {r['full_name'] for r in server}
    remote_scans=[]
    for repo in server:
        if repo['full_name'] not in local_names and repo['name']!=SELF.name:
            try: remote_scans.append(scan_archive(repo))
            except (OSError,RuntimeError,ValueError): gaps.append('Remote archive unavailable: '+repo['full_name'])
    return {'schema_version':1,'forge':{'base_url':BASE,'version':get(BASE+'/api/v1/version'),'authentication':'anonymous'},
            'g1_complete':False, 'coverage_gaps':gaps,
            'server_repositories':sorted(server,key=lambda r:r['id']),
            'registered_repositories':sorted(registered,key=lambda r:r['slug']),
            'packages': sorted([{'id':p.get('id'),'name':p.get('name'),'type':p.get('type'),'version':p.get('version'),
                                'owner':(p.get('owner') or {}).get('login'),
                                'linked_repository_id':(p.get('repository') or {}).get('id')} for p in packages],key=lambda p:(str(p['id']),str(p['version']))),
            'reconciliation':{'server_without_local_origin':sorted(public_names-local_names-{f'coulomb/{SELF.name}'}),
                              'local_origin_not_publicly_listed':sorted(local_names-public_names),
                              'coordinator_excluded':str(SELF)},
            'local_repositories':scans,'remote_archives':remote_scans}


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument('--root',type=Path,default=Path('/home/worsch'))
    ap.add_argument('--output',type=Path,required=True)
    ap.add_argument('--compare',type=Path)
    args=ap.parse_args()
    result=collect(args.root)
    encoded=json.dumps(result,sort_keys=True,indent=2)+'\n'
    args.output.write_text(encoded)
    print(json.dumps({'repos':len(result['server_repositories']), 'local_repos':len(result['local_repositories']),
                      'references':sum(len(r['references']) for r in result['local_repositories']),
                      'sha256':hashlib.sha256(encoded.encode()).hexdigest(),'g1_complete':False}))
    if args.compare:
        same=json.loads(args.compare.read_text())==result
        print('normalized_equal='+str(same).lower())
        return 0 if same else 2
    return 0

if __name__=='__main__':
    raise SystemExit(main())