#!/usr/bin/env python3
"""Read-only anonymous admin access probes against a running RustProxy.
No cookie, password or token is sent; response bodies are never printed.
Only HTTP loopback URLs are accepted. Exit 1 on an unexpected disclosure/status.
This is a focused regression probe, NOT a penetration test or full security audit.
"""
from __future__ import annotations
import argparse
import ipaddress
import json
from pathlib import Path
from urllib.parse import urlsplit
from urllib.request import Request, build_opener, ProxyHandler, HTTPRedirectHandler
from urllib.error import HTTPError, URLError

PRIVATE = ('/api/status','/api/config','/api/backends','/api/traffic','/api/cache','/api/transform','/api/logs','/api/audit',
           '/api/events','/api/certificates','/api/acme','/api/auth/session','/metrics','/api/metrics',
           '/docs/openapi.json','/docs/init.js','/docs/style.css','/docs/swagger-ui-bundle.js',
           '/docs/swagger-ui.css','/docs/licenses.txt','/assets/app.js','/assets/langs.json','/assets/vhosts.js',
           '/assets/management.js','/assets/integrations.js','/assets/settings.js','/assets/acme.js',
           '/assets/traffic-panel.js','/assets/cache-panel.js','/assets/transform-panel.js','/api/does-not-exist')

class NoRedirect(HTTPRedirectHandler):
    def redirect_request(self,*args,**kwargs):
        return None

def base_url(value: str) -> str:
    url=urlsplit(value)
    if url.scheme!='http' or url.username or url.password or url.path not in ('','/') or url.query or url.fragment:
        raise argparse.ArgumentTypeError('Use the HTTP loopback administrator origin, e.g. http://127.0.0.1:9090')
    try:
        loopback=url.hostname=='localhost' or ipaddress.ip_address(url.hostname).is_loopback
        if not loopback or not url.port:raise ValueError()
    except (ValueError,TypeError):raise argparse.ArgumentTypeError('Explicit loopback address and port required')
    return value.rstrip('/')

def main():
    parser=argparse.ArgumentParser(description=__doc__);parser.add_argument('--url',type=base_url,default='http://127.0.0.1:9090');args=parser.parse_args()
    opener=build_opener(ProxyHandler({}),NoRedirect());failed=0;count=0
    def check(path,expected,validate):
        nonlocal failed,count
        count+=1
        try:
            request=Request(args.url+path,headers={'Accept':'application/json'})
            try:response=opener.open(request,timeout=5)
            except HTTPError as error:response=error
            with response:
                status=response.code;headers=response.headers;body=response.read(8388609 if path=='/assets/login-langs.json' else 131073)
            ok=status==expected and headers.get('Cache-Control')=='no-store' and validate(body,headers)
        except (OSError,ValueError,KeyError,TypeError,URLError):status='unavailable';ok=False
        if not ok:failed+=1
        print(f'{"PASS" if ok else "FAIL"} {path}: HTTP {status}; expected {expected}')
    def denied(body,headers):return json.loads(body)=={'error':'auth.required'}
    for path in PRIVATE:check(path,401,denied)
    check('/docs',303,lambda b,h:h.get('Location')=='/?next=docs')
    check('/docs/',303,lambda b,h:h.get('Location')=='/?next=docs')
    check('/healthz',200,lambda b,h:json.loads(b)=={'status':'ok'})
    check('/api/auth/status',200,lambda b,h:isinstance((v:=json.loads(b)).get('setup_required'),bool) and set(v)=={'setup_required'})
    check('/',200,lambda b,h:b'/assets/login.js' in b and all(x not in b for x in [b'/assets/app.js',b'config-editor',b'loginLead']))
    check('/assets/login-langs.json',200,lambda b,h:all(set(p['translations'])<=set(json.loads((Path(__file__).resolve().parents[1]/'src/admin/web/login-keys.json').read_text())) for p in json.loads(b)['languages']))
    print(f'{count-failed}/{count} anonymous probes passed. No authenticated/mutation/TLS checks performed.')
    raise SystemExit(bool(failed))
if __name__=='__main__':main()