v0.34.9_css_refactor

tools/audit/admin_audit.py

@@ -1,63 +0,0 @@
-#!/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[2]/'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()

tools/dev/test_compression.py

@@ -1,260 +0,0 @@
-#!/usr/bin/env python3
-
-import argparse
-import gzip
-import hashlib
-import io
-import time
-import zlib
-
-import brotli
-import requests
-import zstandard
-
-
-TESTS = {
-    "identity": "identity",
-    "gzip": "gzip",
-    "deflate": "deflate",
-    "brotli": "br",
-    "zstd": "zstd",
-    "all": "zstd, br, gzip, deflate",
-}
-
-
-def decompress(data, content_encoding):
-    encodings = [
-        e.strip().lower()
-        for e in content_encoding.split(",")
-        if e.strip() and e.strip().lower() != "identity"
-    ]
-
-    # Content-Encoding usuwa się w odwrotnej kolejności.
-    for encoding in reversed(encodings):
-        if encoding == "gzip":
-            data = gzip.decompress(data)
-
-        elif encoding == "deflate":
-            try:
-                data = zlib.decompress(data)
-            except zlib.error:
-                # Niektóre serwery wysyłają raw DEFLATE.
-                data = zlib.decompress(data, -zlib.MAX_WBITS)
-
-        elif encoding == "br":
-            data = brotli.decompress(data)
-
-        elif encoding == "zstd":
-            dctx = zstandard.ZstdDecompressor()
-            with dctx.stream_reader(io.BytesIO(data)) as reader:
-                data = reader.read()
-
-        else:
-            raise RuntimeError(f"Nieznany Content-Encoding: {encoding}")
-
-    return data
-
-
-def human_size(size):
-    if size < 1024:
-        return f"{size} B"
-    if size < 1024 * 1024:
-        return f"{size / 1024:.1f} KB"
-    return f"{size / 1024 / 1024:.2f} MB"
-
-
-def request_test(url, accept_encoding, verify_ssl=True, timeout=15, revalidate=False):
-    headers = {
-        "Accept-Encoding": accept_encoding,
-        "User-Agent": "compression-tester/1.0",
-    }
-    if revalidate:
-        headers["Cache-Control"] = "no-cache"
-
-    start = time.perf_counter()
-
-    r = requests.get(
-        url,
-        headers=headers,
-        timeout=timeout,
-        verify=verify_ssl,
-        stream=True,
-        allow_redirects=True,
-    )
-
-    r.raw.decode_content = False
-    raw = r.raw.read()
-
-    elapsed = time.perf_counter() - start
-
-    content_encoding = r.headers.get("Content-Encoding", "identity")
-
-    try:
-        decoded = decompress(raw, content_encoding)
-        error = None
-    except Exception as exc:
-        decoded = b""
-        error = str(exc)
-
-    return {
-        "status": r.status_code,
-        "requested": accept_encoding,
-        "received": content_encoding,
-        "raw_size": len(raw),
-        "decoded_size": len(decoded),
-        "time": elapsed,
-        "decoded": decoded,
-        "error": error,
-        "headers": r.headers,
-        "final_url": r.url,
-    }
-
-
-def main():
-    parser = argparse.ArgumentParser(
-        description="Tester HTTP Content-Encoding: zstd/br/gzip/deflate"
-    )
-
-    parser.add_argument("url", help="URL do przetestowania")
-    parser.add_argument(
-        "--insecure",
-        action="store_true",
-        help="Wyłącz weryfikację certyfikatu SSL",
-    )
-    parser.add_argument(
-        "--timeout",
-        type=int,
-        default=15,
-        help="Timeout HTTP w sekundach",
-    )
-    parser.add_argument(
-        "--revalidate",
-        action="store_true",
-        help="Wyslij Cache-Control: no-cache (wymusza rewalidacje i moze resetowac Age do 0)",
-    )
-
-    args = parser.parse_args()
-
-    if args.insecure:
-        requests.packages.urllib3.disable_warnings()
-
-    print()
-    print(f"URL: {args.url}")
-    print("=" * 100)
-
-    results = {}
-
-    for name, accept_encoding in TESTS.items():
-        try:
-            result = request_test(
-                args.url,
-                accept_encoding,
-                verify_ssl=not args.insecure,
-                timeout=args.timeout,
-                revalidate=args.revalidate,
-            )
-            results[name] = result
-        except Exception as exc:
-            results[name] = {
-                "error": f"REQUEST ERROR: {exc}"
-            }
-
-    identity_body = None
-
-    if "identity" in results:
-        identity_body = results["identity"].get("decoded")
-
-    print(
-        f"{'TEST':<10}"
-        f"{'HTTP':<7}"
-        f"{'REQUEST':<27}"
-        f"{'RESPONSE':<12}"
-        f"{'RAW':<12}"
-        f"{'DECODED':<12}"
-        f"{'RATIO':<10}"
-        f"{'TIME':<10}"
-        f"{'BODY'}"
-    )
-
-    print("-" * 120)
-
-    failed = False
-
-    for name, result in results.items():
-        if result.get("error"):
-            print(f"{name:<10} ERROR: {result['error']}")
-            failed = True
-            continue
-
-        raw_size = result["raw_size"]
-        decoded_size = result["decoded_size"]
-
-        if decoded_size:
-            ratio = raw_size / decoded_size
-            ratio_text = f"{ratio * 100:.1f}%"
-        else:
-            ratio_text = "-"
-
-        if identity_body is not None and result["decoded"]:
-            if result["decoded"] == identity_body:
-                body_status = "OK"
-            else:
-                body_status = "DIFF"
-        else:
-            body_status = "-"
-
-        print(
-            f"{name:<10}"
-            f"{result['status']:<7}"
-            f"{result['requested']:<27}"
-            f"{result['received']:<12}"
-            f"{human_size(raw_size):<12}"
-            f"{human_size(decoded_size):<12}"
-            f"{ratio_text:<10}"
-            f"{result['time']:.3f}s    "
-            f"{body_status}"
-        )
-
-    print()
-    print("=" * 100)
-    print("Szczegóły nagłówków:")
-    print()
-
-    for name, result in results.items():
-        if result.get("error"):
-            continue
-
-        headers = result["headers"]
-
-        print(f"[{name}]")
-        print(f"  URL finalny:       {result['final_url']}")
-        print(f"  HTTP status:       {result['status']}")
-        print(f"  Content-Encoding:  {headers.get('Content-Encoding', '-')}")
-        print(f"  Content-Type:      {headers.get('Content-Type', '-')}")
-        print(f"  Content-Length:    {headers.get('Content-Length', '-')}")
-        print(f"  Vary:              {headers.get('Vary', '-')}")
-        print(f"  Cache-Control:     {headers.get('Cache-Control', '-')}")
-        print(f"  Age:               {headers.get('Age', '-')}")
-        print(f"  X-Cache:           {headers.get('X-Cache', '-')}")
-        print(f"  X-Cache-Hits:      {headers.get('X-Cache-Hits', '-')}")
-        print(f"  Cache-Status:      {headers.get('Cache-Status', '-')}")
-        print(f"  ETag:              {headers.get('ETag', '-')}")
-        print(f"  Server:            {headers.get('Server', '-')}")
-        print()
-
-    print("SHA256 po dekompresji:")
-    print()
-
-    for name, result in results.items():
-        body = result.get("decoded")
-
-        if body:
-            sha = hashlib.sha256(body).hexdigest()
-            print(f"{name:<10} {sha}")
-
-    if failed:
-        raise SystemExit(1)
-
-
-if __name__ == "__main__":
-    main()