#!/usr/bin/env python3
"""Security self-check: probe files through an nginx-like proxy, cleanup, cache, findings and API access."""
import argparse
import fcntl
import functools
import http.client
import http.server
import json
import os
from pathlib import Path
import re
import shutil
import socket
import ssl
import subprocess
import tempfile
import threading
import time
import traceback
import urllib.parse
from support import Client, Site

# Extensions that HestiaCP's nginx serves from disk without asking Apache.
STATIC = {".json", ".txt", ".zip", ".jpg", ".mp3", ".mp4", ".pdf", ".css", ".js", ".svg", ".webp", ".png"}


def front_proxy(root, backend_port, deny=(), tls=None):
    """Serve existing static files itself, deny prefixes like `location ^~`, proxy the rest to PHP."""
    class Handler(http.server.BaseHTTPRequestHandler):
        def log_message(self, *args):
            pass

        def do_GET(self):
            path = urllib.parse.unquote(urllib.parse.urlsplit(self.path).path)
            if any(path.startswith(prefix) for prefix in deny):
                return self.reply(403, b"denied by front proxy")
            file = Path(root) / path.lstrip("/")
            if Path(path).suffix.lower() in STATIC and file.is_file():
                return self.reply(200, file.read_bytes())
            backend = http.client.HTTPConnection("127.0.0.1", backend_port, timeout=10)
            backend.request("GET", self.path, headers={"Host": self.headers.get("Host", "")})
            response = backend.getresponse()
            self.reply(response.status, response.read(), response.getheader("Location"))

        def reply(self, status, body, location=None):
            self.send_response(status)
            if location:
                self.send_header("Location", location)
            self.send_header("Content-Length", str(len(body)))
            self.end_headers()
            self.wfile.write(body)

    server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler)
    if tls:
        server.socket = tls.wrap_socket(server.socket, server_side=True)
    threading.Thread(target=server.serve_forever, daemon=True).start()
    return server


def redirect_server(target):
    class Handler(http.server.BaseHTTPRequestHandler):
        def log_message(self, *args):
            pass

        def do_GET(self):
            self.send_response(301)
            self.send_header("Location", target + self.path)
            self.send_header("Content-Length", "0")
            self.end_headers()

    server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler)
    threading.Thread(target=server.serve_forever, daemon=True).start()
    return server


class QuietStaticHandler(http.server.SimpleHTTPRequestHandler):
    def log_message(self, *args):
        pass


def php(root, code, server=None, constants=None):
    """Run the helper in a fresh PHP process; any notice or warning fails the test."""
    prelude = "$_SERVER = array_merge($_SERVER, json_decode(getenv('RJ_SERVER'), true));"
    for name, value in (constants or {}).items():
        prelude += f"define({name!r}, {json.dumps(value)});"
    prelude += "require 'includes/security-check.php';"
    result = subprocess.run(["php", "-d", "display_errors=stderr", "-d", "error_reporting=-1", "-r", prelude + code],
                            cwd=root, capture_output=True, text=True,
                            env=dict(os.environ, RJ_SERVER=json.dumps({"SCRIPT_NAME": "/admin/api.php", **(server or {})})))
    assert result.returncode == 0 and not result.stderr, result.stderr + result.stdout
    return json.loads(result.stdout)


def probe(root, base):
    return php(root, f"echo json_encode(raijuSecurityRunProbe(getcwd(), {json.dumps(base)}));")


def leftovers(root):
    return [str(path) for path in Path(root).rglob("raiju-access-check-*")]


def items(status):
    return {item["id"]: item for item in status["items"]}


def free_port():
    with socket.socket() as sock:
        sock.bind(("127.0.0.1", 0))
        return sock.getsockname()[1]


def test_probe_through_proxy(site):
    root = site.root
    folders = ["content/", "backups/", "assets/images-trash/", "assets/audio-trash/", "assets/videos-trash/", "assets/documents-trash/"]
    exposed_proxy = front_proxy(root, site.port)
    try:
        result = probe(root, f"http://127.0.0.1:{exposed_proxy.server_port}")
    finally:
        exposed_proxy.shutdown()
    assert result["state"] == "exposed", result
    assert [p["path"] for p in result["probes"]] == folders, result["probes"]
    assert all(p["status"] == 200 and p["state"] == "exposed" for p in result["probes"]), result["probes"]
    assert result["http"] == "skipped", "Plain http:// base must not run the redirect check"
    assert not leftovers(root), leftovers(root)

    # The generated nginx rules must cover every probed folder.
    rules = php(root, "echo json_encode(raijuSecurityNginxRules(''));")
    prefixes = re.findall(r"^location \^~ (\S+) \{ deny all; \}$", rules, re.M)
    assert len(prefixes) == 8 and all(("/" + folder) in prefixes for folder in folders), rules
    assert "location ^~ /sub/content/ { deny all; }" in php(root, "echo json_encode(raijuSecurityNginxRules('/sub'));")
    fixed_proxy = front_proxy(root, site.port, deny=prefixes)
    try:
        result = probe(root, f"http://127.0.0.1:{fixed_proxy.server_port}")
    finally:
        fixed_proxy.shutdown()
    assert result["state"] == "protected" and all(p["status"] == 403 for p in result["probes"]), result
    # A proxy that still serves media itself (e.g. a Hestia template limited to assets) exposes only the trash folders.
    media_proxy = front_proxy(root, site.port, deny=["/content/", "/backups/"])
    try:
        result = probe(root, f"http://127.0.0.1:{media_proxy.server_port}")
    finally:
        media_proxy.shutdown()
    assert [p["path"] for p in result["probes"] if p["state"] == "exposed"] == folders[2:], result
    # Without a proxy, .htaccess (Apache) or router.php (development server) blocks the folders.
    result = probe(root, f"http://127.0.0.1:{site.port}")
    assert result["state"] == "protected", result
    assert not leftovers(root), leftovers(root)


def test_server_ignoring_htaccess(site):
    # A bare PHP server resembles nginx + PHP-FPM without rules: files and CLI scripts are reachable.
    port = free_port()
    bare = subprocess.Popen(["php", "-d", "display_errors=0", "-S", f"127.0.0.1:{port}", "-t", str(site.root)],
                            stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    try:
        for _ in range(100):
            try:
                with socket.create_connection(("127.0.0.1", port), timeout=.1):
                    break
            except OSError:
                time.sleep(.05)
        result = probe(site.root, f"http://127.0.0.1:{port}")
        assert result["state"] == "exposed" and "content/" in [p["path"] for p in result["probes"] if p["state"] == "exposed"], result
        for name in ("backup", "make", "convert"):
            connection = http.client.HTTPConnection("127.0.0.1", port, timeout=10)
            connection.request("GET", f"/cli/{name}.php?x+--action=run")
            response = connection.getresponse()
            body = response.read()
            # Web SAPIs echo a shebang line as text; nothing else may run or be printed.
            assert response.status == 404 and body.strip() in (b"", b"#!/usr/bin/env php"), \
                f"cli/{name}.php ran as a web request: {response.status} {body[:200]!r}"
        assert not list((site.root / "backups").glob("*-backup-*.zip")), "Web request created a backup"
    finally:
        bare.terminate()
        bare.wait(timeout=10)
    assert not leftovers(site.root), leftovers(site.root)


def test_https_redirect_and_self_signed(site):
    if not shutil.which("openssl"):
        print("SKIP test_https_redirect_and_self_signed: openssl is not installed")
        return
    root = site.root
    prefixes = re.findall(r"^location \^~ (\S+) \{ deny all; \}$", php(root, "echo json_encode(raijuSecurityNginxRules(''));"), re.M)
    with tempfile.TemporaryDirectory(prefix="raiju-tls-") as folder:
        cert, key = Path(folder) / "cert.pem", Path(folder) / "key.pem"
        subprocess.run(["openssl", "req", "-x509", "-newkey", "rsa:2048", "-nodes", "-days", "1", "-subj", "/CN=127.0.0.1",
                        "-keyout", str(key), "-out", str(cert)], check=True, capture_output=True)
        context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
        context.load_cert_chain(cert, key)
        # Staging servers often use self-signed certificates: the status-only probe still works.
        secure = front_proxy(root, site.port, deny=prefixes, tls=context)
        base = f"https://127.0.0.1:{secure.server_port}"
        redirect = redirect_server(base)
        plain = front_proxy(root, site.port, deny=prefixes)
        try:
            def run(plain_base):
                return php(root, f"echo json_encode(raijuSecurityRunProbe(getcwd(), {json.dumps(base)}, {json.dumps(plain_base)}));")
            result = run(f"http://127.0.0.1:{redirect.server_port}")
            assert result["state"] == "protected" and result["http"] == "redirect", result
            assert run(f"http://127.0.0.1:{plain.server_port}")["http"] == "open"
            assert run(f"http://127.0.0.1:{free_port()}")["http"] == "closed"
            # An https:// address with its own port has no known http:// counterpart.
            assert probe(root, base)["http"] == "skipped"
        finally:
            for server in (secure, redirect, plain):
                server.shutdown()
    assert not leftovers(root), leftovers(root)


def test_wrong_address_and_cleanup(site):
    root = site.root
    stale = root / "content/raiju-access-check-0123456789abcdef.json"
    unrelated = root / "content/raiju-access-check-notes.json"
    stale.write_text("{}")
    unrelated.write_text("{}")
    # Another installation answers the address: its 404 for the control file must not count as protected.
    with tempfile.TemporaryDirectory(prefix="raiju-other-") as other:
        (Path(other) / "content").mkdir()
        handler = functools.partial(QuietStaticHandler, directory=other)
        other_server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), handler)
        threading.Thread(target=other_server.serve_forever, daemon=True).start()
        try:
            result = probe(root, f"http://127.0.0.1:{other_server.server_port}")
        finally:
            other_server.shutdown()
    assert result["state"] == "unknown" and result["reason"] == "unreachable" and result["detail"] == "HTTP 404", result
    assert not stale.exists(), "Stale probe file was not swept"
    assert unrelated.exists(), "Sweep removed a file that is not a probe"
    unrelated.unlink()
    started = time.monotonic()
    result = probe(root, f"http://127.0.0.1:{free_port()}")
    assert result["state"] == "unknown" and result["reason"] == "unreachable" and result["detail"], result
    assert time.monotonic() - started < 10, "Unreachable address was not answered quickly"
    assert not leftovers(root), leftovers(root)


def test_findings():
    with tempfile.TemporaryDirectory(prefix="raiju-security-findings-") as folder:
        root = Site(folder).root
        public = {"HTTP_HOST": "www.example.com", "HTTPS": "on", "REMOTE_ADDR": "203.0.113.9", "SERVER_ADDR": "198.51.100.1"}
        status = php(root, "echo json_encode(raijuSecurityStatus('https'));", public,
                     {"RAIJU_INITIAL_DISPLAY_ERRORS": "0", "RAIJU_DEV_LOGIN": False})
        found = items(status)
        assert not status["local"] and not status["devServer"]
        assert found["folders"]["state"] == "pending" and found["folders"]["stale"], found["folders"]
        assert found["http"]["state"] == "pending", found["http"]
        assert all(found[key]["state"] == "ok" for key in ("https", "cookie", "client_ip", "errors", "dev_login")), found
        assert found["php"]["version"] and found["php"]["state"] in ("ok", "warning")

        # Proxy without visitor IPs, unrecognised HTTPS, visible errors and the default development login.
        found = items(php(root, "echo json_encode(raijuSecurityStatus('https'));",
                          {**public, "HTTPS": "", "REMOTE_ADDR": "127.0.0.1"}, {"RAIJU_INITIAL_DISPLAY_ERRORS": "On"}))
        assert found["cookie"]["state"] == "warning" and found["errors"]["state"] == "warning", found
        assert found["client_ip"] == {"id": "client_ip", "state": "warning", "ip": "127.0.0.1"}, found["client_ip"]
        assert found["dev_login"]["state"] == "warning", found["dev_login"]
        found = items(php(root, "echo json_encode(raijuSecurityStatus('https'));",
                          {**public, "REMOTE_ADDR": "198.51.100.1"}, {"RAIJU_INITIAL_DISPLAY_ERRORS": "stderr"}))
        assert found["client_ip"]["state"] == "warning" and found["dev_login"]["state"] == "note", found
        assert found["errors"]["state"] == "ok", "display_errors=stderr does not reach visitors"

        found = items(php(root, "echo json_encode(raijuSecurityStatus('http'));", {**public, "HTTPS": ""},
                          {"RAIJU_INITIAL_DISPLAY_ERRORS": "stdout"}))
        assert found["https"]["state"] == "warning" and found["errors"]["state"] == "warning", found
        assert found["cookie"] == {"id": "cookie", "state": "skipped", "reason": "scheme"}, found["cookie"]
        assert found["http"] == {"id": "http", "state": "skipped", "reason": "scheme"}, found["http"]

        status = php(root, "echo json_encode(raijuSecurityStatus('http'));", {"HTTP_HOST": "localhost:3000", "REMOTE_ADDR": "127.0.0.1"})
        found = items(status)
        assert status["local"] and found["folders"]["state"] == "skipped" and found["folders"]["reason"] == "local", found
        assert all(found[key]["state"] == "skipped" for key in ("https", "cookie", "client_ip", "errors", "dev_login")), found

        found = items(php(root, "echo json_encode(raijuSecurityStatus('https'));", {**public, "HTTP_HOST": "evil.example/x"}))
        assert found["folders"]["state"] == "unknown" and found["folders"]["reason"] == "host", found["folders"]

        hosts = php(root, "echo json_encode(array_map('raijuSecurityIsLocalHost', ['localhost:3000', '127.0.0.1', '[::1]:8080', "
                          "'site.test', 'raiju.localhost', '192.168.1.20', 'intranet', 'www.example.com', '93.184.215.14:8443', "
                          "'beruehrungwirkt.mybyte.at']));")
        assert hosts == [True] * 7 + [False] * 3, hosts


def test_cache_and_lock():
    with tempfile.TemporaryDirectory(prefix="raiju-security-cache-") as folder:
        root = Site(folder).root
        cache = root / "content/security-check.json"
        public = {"HTTP_HOST": "www.example.com", "HTTPS": "on", "REMOTE_ADDR": "203.0.113.9"}
        entry = {"version": 1, "baseUrl": "https://www.example.com", "checkedAt": "2026-09-26T10:00:00+02:00",
                 "checkedTs": int(time.time()) - 60, "state": "exposed", "reason": "", "detail": "",
                 "probes": [{"path": "content/", "status": 200, "state": "exposed"}, {"path": "backups/", "status": 403, "state": "protected"}],
                 "http": "open"}
        cache.write_text(json.dumps(entry))
        # A recent result is reused without any network request, even when running is allowed.
        found = items(php(root, "echo json_encode(raijuSecurityStatus('https', true));", public))
        assert found["folders"]["state"] == "critical" and found["folders"]["exposed"] == ["content/"], found["folders"]
        assert not found["folders"]["stale"] and "location ^~ /content/ { deny all; }" in found["folders"]["rules"]
        assert found["http"] == {"id": "http", "state": "warning", "mode": "open"}, found["http"]
        assert json.loads(cache.read_text())["checkedTs"] == entry["checkedTs"], "Fresh cache was re-run"
        # Deleted media alone is a warning (no dashboard banner), still with the nginx rules.
        trash = {**entry, "probes": [{"path": "content/", "status": 403, "state": "protected"},
                                     {"path": "assets/images-trash/", "status": 200, "state": "exposed"}]}
        cache.write_text(json.dumps(trash))
        found = items(php(root, "echo json_encode(raijuSecurityStatus('https'));", public))["folders"]
        assert found["state"] == "warning" and found["exposed"] == ["assets/images-trash/"] and found["rules"], found
        cache.write_text(json.dumps(entry))
        # Exposed results expire after ten minutes; protected ones after twelve hours.
        cache.write_text(json.dumps({**entry, "checkedTs": int(time.time()) - 601}))
        assert items(php(root, "echo json_encode(raijuSecurityStatus('https'));", public))["folders"]["stale"]
        cache.write_text(json.dumps({**entry, "state": "protected", "checkedTs": int(time.time()) - 601}))
        assert not items(php(root, "echo json_encode(raijuSecurityStatus('https'));", public))["folders"]["stale"]
        # Another address (moved site, restored backup) must not inherit the result.
        assert items(php(root, "echo json_encode(raijuSecurityStatus('https'));", {**public, "HTTP_HOST": "staging.example.com"}))["folders"]["state"] == "pending"
        # A running check holds the lock: report it instead of starting a second probe.
        cache.write_text(json.dumps(entry))
        with open(str(cache) + ".lock", "a") as lock:
            fcntl.flock(lock, fcntl.LOCK_EX)
            found = items(php(root, "echo json_encode(raijuSecurityStatus('https', true, true));", public))
            fcntl.flock(lock, fcntl.LOCK_UN)
        assert found["folders"]["running"] and found["folders"]["state"] == "critical", found["folders"]


def test_api(site):
    guest = Client(site)
    assert guest.api("security-check")[0] == 401
    admin = Client(site).login()
    editor = Client(site).login("editor")
    assert editor.api("security-check")[0] == 403
    # Only administrators get the warning banner and the system status badge.
    for client, visible in ((admin, True), (editor, False)):
        body = client.request("/admin/dashboard.php")[2]
        assert (b'id="securityExposureWarning"' in body and b'id="systemBadge"' in body) == visible, client
    status, _, _ = admin.request("/admin/api.php?action=security-check")
    assert status == 405, "Probe run must require POST"
    assert admin.api("security-check", csrf_token="wrong")[0] == 403
    apache = hasattr(site, "fpm_process")
    if apache:
        # A local address is only probed on request, then through Apache and the real .htaccess.
        status, result = admin.api("security-check", scheme="http")
        assert status == 200 and items(result["data"])["folders"]["state"] == "skipped", result
    status, result = admin.api("security-check", scheme="http", force="1")
    assert status == 200 and result["success"], result
    found = items(result["data"])
    if apache:
        assert found["folders"]["state"] == "ok" and all(p["status"] == 403 for p in found["folders"]["probes"]), found["folders"]
        assert json.loads((site.root / "content/security-check.json").read_text())["state"] == "protected"
    else:
        # The development server is single-threaded and cannot request itself.
        assert found["folders"] == {"id": "folders", "canRun": False, "url": f"http://127.0.0.1:{site.port}",
                                    "state": "skipped", "reason": "dev_server"}, found["folders"]
    assert found["http"]["state"] == "skipped" and found["php"]["version"], found
    status, _, body = admin.request("/admin/api.php?action=system-status&scheme=http")
    security = json.loads(body)["data"]["security"]
    assert status == 200 and security["devServer"] != apache, security
    assert items(security)["folders"]["state"] == ("ok" if apache else "skipped"), security
    assert not leftovers(site.root)


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--apache", action="store_true", help="use Apache/PHP-FPM with the real .htaccess behind the proxy")
    args = parser.parse_args()
    if args.apache:
        from apache_support import ApacheSite as Backend
    else:
        Backend = Site
    failures = []
    for test in (test_findings, test_cache_and_lock):
        try:
            test()
            print("PASS", test.__name__)
        except Exception:
            failures.append(test.__name__)
            traceback.print_exc()
    with tempfile.TemporaryDirectory(prefix="raiju-security-smoke-") as folder:
        site = Backend(folder)
        try:
            site.start()
            if args.apache:
                print(site.server_version + "; " + site.php_version)
            for test in (test_probe_through_proxy, test_server_ignoring_htaccess, test_https_redirect_and_self_signed,
                         test_wrong_address_and_cleanup, test_api):
                try:
                    test(site)
                    print("PASS", test.__name__)
                except Exception:
                    failures.append(test.__name__)
                    traceback.print_exc()
        finally:
            site.close()
    if failures:
        raise SystemExit("Failed: " + ", ".join(failures))