security: verify forge scanner expansion across 166 repositories

docs/changes/CUST-WP-0080/probe_forge_api_access.py

@@ -0,0 +1,69 @@
+#!/usr/bin/env python3
+"""Silent read-only API admission through the existing protected credential route."""
+import argparse
+import datetime
+import hashlib
+import json
+import os
+from pathlib import Path
+import subprocess
+import sys
+import urllib.error
+import urllib.request
+
+URL = 'https://forgejo.coulomb.social/api/v1/user'
+
+
+class NoRedirect(urllib.request.HTTPRedirectHandler):
+    def redirect_request(self, req, fp, code, msg, headers, newurl):
+        return None
+
+
+def inner(receipt):
+    result={'scope':'GET /api/v1/user only; no repository, identity or credential changes',
+            'at':datetime.datetime.now(datetime.timezone.utc).isoformat(),'result':'unverified'}
+    token=os.environ.get('API_TOKEN','')
+    if not token:
+        result['reason']='protected-injection-absent'
+    else:
+        try:
+            opener=urllib.request.build_opener(urllib.request.ProxyHandler({}),NoRedirect())
+            request=urllib.request.Request(URL,headers={'Authorization':'token '+token,'Accept':'application/json'})
+            with opener.open(request,timeout=15) as response:
+                result['http_status']=response.status
+                body=json.loads(response.read(65536))
+            result['expected_caller']=body.get('login')=='tegwick'
+            result['caller_id']=body.get('id') if isinstance(body.get('id'),int) else None
+            result['site_admin']=body.get('is_admin') is True
+            result['result']='authenticated' if result['http_status']==200 and result['expected_caller'] else 'unverified'
+        except urllib.error.HTTPError as error:
+            result.update(http_status=error.code,reason='http-refused')
+        except Exception:
+            result['reason']='request-unverified'
+    receipt.write_text(json.dumps(result,indent=2)+'\n')
+    return 0 if result['result']=='authenticated' else 1
+
+
+def main():
+    parser=argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('--receipt',type=Path,required=True)
+    parser.add_argument('--inner',action='store_true')
+    args=parser.parse_args()
+    if args.inner:return inner(args.receipt)
+    if args.receipt.exists():raise SystemExit('Receipt already exists; retain the prior attempt.')
+    env=dict(os.environ,WARDEN_AGENT_ID='custodian-codex')
+    result=subprocess.run(['warden','access','forgejo-admin-api-token','--field','API_TOKEN','--exec','--',
+                           sys.executable,str(Path(__file__).resolve()),'--inner','--receipt',str(args.receipt.resolve())],
+                          capture_output=True,env=env,timeout=90)
+    if args.receipt.exists():record=json.loads(args.receipt.read_text())
+    else:record={'result':'route-unavailable','scope':'Protected read-only API admission; child receipt absent'}
+    record.update(route_exit_code=result.returncode,executor_sha256=hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),
+                  token_returned_to_agent=False,raw_child_diagnostics_retained=False)
+    args.receipt.write_text(json.dumps(record,indent=2)+'\n')
+    print(json.dumps(record))
+    return 0 if record['result']=='authenticated' else 1
+
+
+if __name__=='__main__':
+    try:sys.exit(main())
+    except Exception:sys.exit('Protected API admission incomplete; raw diagnostics suppressed.')