64 lines
3.9 KiB
Python
Executable File
64 lines
3.9 KiB
Python
Executable File
#!/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/file-logs','/api/file-logs/query','/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/file-logs.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()
|