wordpress_check.py

#!/usr/bin/env python3
# SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-CompleteTech-Commercial
# Copyright (c) 2026 CompleteTech LLC. See LICENSING.md.
"""Passive PHP2Shell exposure checks and WordPress detection. Python 3.10+."""

from __future__ import annotations

import argparse
import base64
import csv
import html
import hashlib
import io
import os
import stat
import tempfile
import json
import math
import mimetypes
import re
import socket
import subprocess
import sys
import time
from dataclasses import asdict, dataclass, field
from datetime import datetime, timezone
from html.parser import HTMLParser
from http.client import HTTPException
from pathlib import Path
from urllib.error import HTTPError, URLError
from urllib.parse import parse_qs, quote, urljoin, urlsplit, urlunsplit
from urllib.request import HTTPRedirectHandler, Request, build_opener


TOOL_NAME = "php2shell-check"
TOOL_VERSION = "1.0.1"
BOT = "WordPressPublicChecker"
REDIRECTS = {301, 302, 303, 307, 308}
MAX_BODY = 2 * 1024 * 1024
MAX_ROBOTS = 512 * 1024
PHP2SHELL_RULES_DATE = "2026-10-21"
PHP2SHELL_CVES = ("CVE-2026-109618", "CVE-2026-109619")
PHP2SHELL_FIXES = {(8, 2): 35, (8, 3): 36, (8, 4): 27, (8, 5): 12}
PHP2SHELL_SOURCES = (
    "https://github.com/php/php-src/security/advisories/GHSA-w8hw-w9fx-j79w",
    "https://github.com/php/php-src/security/advisories/GHSA-6wmf-4rv5-p3rp",
    "https://ubuntu.com/security/notices/USN-8948-1",
)


def assess_php2shell(php_version="", *, version_candidates=None, php_hint=False):
    """Assess advertised upstream release numbers, never exploitability.

    This is offline: no HTTP requests, probes, or payloads are sent. The fixed
    release table is a dated snapshot, not a live vulnerability feed.
    """
    values = version_candidates if version_candidates is not None else [php_version]
    versions = sorted({str(value).strip() for value in values if value and str(value).strip()})
    result = {
        "php2shell_status": "unknown",
        "php2shell_version_status": "version_not_disclosed",
        "php2shell_fixed_version": "",
        "php2shell_exploitability": "not_tested",
        "php2shell_cves": list(PHP2SHELL_CVES),
        "php2shell_rules_as_of": PHP2SHELL_RULES_DATE,
        "php2shell_sources": list(PHP2SHELL_SOURCES),
        "php2shell_notes": [
            "Passive metadata only. Runtime SAPI, vendor backports, configuration, and exploitability are not verified."
        ],
        "php2shell_recommendation": "Ask the site administrator to verify the deployed PHP package and applicable vendor security updates.",
    }
    if not versions:
        result["php2shell_notes"].append(
            "A PHP-related indicator was observed, but no complete PHP version was disclosed."
            if php_hint else "No public PHP version was disclosed; lack of a header does not establish that PHP is absent."
        )
        return result
    if len(versions) > 1:
        result["php2shell_version_status"] = "conflicting_versions"
        result["php2shell_notes"].append("Conflicting PHP versions: " + ", ".join(versions))
        return result
    version = versions[0]
    match = None if len(version) > 128 else re.fullmatch(r"(\d+)\.(\d+)\.(\d+)([A-Za-z0-9.+_~:-]*)", version)
    if not match:
        result["php2shell_version_status"] = "unrecognized_or_partial_version"
        result["php2shell_notes"].append(f"Cannot safely compare PHP version {version[:128]!r} with the upstream release table.")
        return result
    major, minor, patch = (int(match.group(i)) for i in (1, 2, 3))
    suffix = match.group(4)
    fixed_patch = PHP2SHELL_FIXES.get((major, minor))
    if fixed_patch is None:
        result["php2shell_version_status"] = "branch_requires_vendor_review"
        result["php2shell_notes"].append(
            f"PHP {major}.{minor} is outside this rule snapshot's upstream 8.2-8.5 fix table. No safety conclusion is made."
        )
        return result
    fixed = f"{major}.{minor}.{fixed_patch}"
    result["php2shell_fixed_version"] = fixed
    if suffix:
        prerelease = re.search(r"(?:^|[-+._~])(?:dev|alpha|beta|rc)(?:\d|[-._]|$)", suffix, re.I)
        result["php2shell_version_status"] = "prerelease_unverified" if prerelease else "vendor_build_unverified"
        result["php2shell_notes"].append(
            f"PHP {version} contains a build qualifier. Confirm the exact vendor package; its upstream base alone cannot establish patch status."
        )
        return result
    if patch < fixed_patch:
        result["php2shell_status"] = "potentially_affected"
        result["php2shell_version_status"] = "below_upstream_fix"
        result["php2shell_notes"].append(f"Advertised PHP {version} is older than the {fixed} upstream fix for both listed CVEs.")
        result["php2shell_recommendation"] = (
            f"Prioritize administrator review. Verify vendor backports and the actual runtime; install the vendor security update or upstream PHP {fixed} or newer in this branch."
        )
    else:
        result["php2shell_status"] = "fixed_version_reported"
        result["php2shell_version_status"] = "at_or_above_upstream_fix"
        result["php2shell_notes"].append(f"Advertised PHP {version} is at or above the {fixed} upstream fix. This is not a verified clean bill of health.")
        result["php2shell_recommendation"] = "Confirm that the advertised version is the deployed, patched runtime and follow the current vendor advisory."
    return result


def annotate_php2shell(result):
    assessment = assess_php2shell(
        result.php_version,
        version_candidates=result.php_version_candidates or None,
        php_hint=bool(result.php_evidence) or result.status in {"wordpress", "likely_wordpress"},
    )
    for name, value in assessment.items():
        setattr(result, name, value)


class CheckStopped(Exception):
    """The checker cannot safely or usefully continue."""


class OutOfScope(CheckStopped):
    """A redirect destination lacks explicit origin authorization."""


class TargetDeadline(CheckStopped):
    """The shared monotonic target budget expired."""


class NoRedirect(HTTPRedirectHandler):
    def redirect_request(self, req, fp, code, msg, headers, newurl):
        return None


def robots_path(value: str) -> str:
    """Compare encoded reserved characters literally, decoding unreserved bytes."""
    value = quote(value, safe="/%!$&'()*+,-.:;=?@_~")
    def decode(match):
        char = chr(int(match.group(1), 16))
        return char if char.isascii() and (char.isalnum() or char in "-._~") else match.group(0).upper()
    return re.sub(r"%([0-9a-fA-F]{2})", decode, value)


class RobotsPolicy:
    """Parse user-agent groups, wildcard/anchor rules and common pacing hints."""

    def __init__(self, body: str, minimum_delay: float):
        groups = []
        agents, directives = [], []
        for line in body.lstrip("\ufeff").splitlines():
            line = line.split("#", 1)[0].strip()
            if ":" not in line:
                continue
            key, value = (part.strip() for part in line.split(":", 1))
            key = key.lower()
            if key == "user-agent":
                if directives:
                    groups.append((agents, directives))
                    agents, directives = [], []
                agents.append(value.lower())
            elif agents:
                directives.append((key, value))
        if agents:
            groups.append((agents, directives))
        selected = []
        best = -1
        for agents, directives in groups:
            specificity = max((0 if a == "*" else len(a) for a in agents if a and (a == "*" or a in BOT.lower())), default=-1)
            if specificity < 0:
                continue
            if specificity > best:
                selected, best = [], specificity
            if specificity == best:
                selected.extend(directives)
        self.rules = []
        self.delay = minimum_delay
        for key, value in selected:
            if key in {"allow", "disallow"} and value:
                path = robots_path(value)
                anchored = path.endswith("$")
                if anchored:
                    path = path[:-1]
                pattern = "^" + ".*".join(re.escape(part) for part in path.split("*")) + ("$" if anchored else "")
                self.rules.append((re.compile(pattern), len(path.replace("*", "").encode("utf-8")), key == "allow"))
            elif key in {"crawl-delay", "request-rate"}:
                try:
                    if key == "crawl-delay":
                        delay = float(value)
                    else:
                        count, seconds = map(float, value.split("/"))
                        delay = seconds / count
                    if math.isfinite(delay) and delay >= 0:
                        self.delay = max(self.delay, delay)
                except (ValueError, ZeroDivisionError):
                    pass

    def can_fetch(self, url: str) -> bool:
        parts = urlsplit(url)
        path = robots_path(parts.path or "/")
        if parts.query:
            path += "?" + robots_path(parts.query)
        matches = [(size, allow) for regex, size, allow in self.rules if regex.search(path)]
        return max(matches)[1] if matches else True


def normalize_url(value: str) -> str:
    value = value.strip()
    if not value or any(c.isspace() or ord(c) < 32 for c in value):
        raise ValueError("URL must not be empty or contain whitespace")
    if "://" not in value:
        value = "https://" + value
    parts = urlsplit(value)
    if parts.scheme.lower() not in {"http", "https"} or not parts.hostname:
        raise ValueError("Only HTTP(S) website URLs are supported")
    if parts.username is not None or parts.password is not None:
        raise ValueError("URLs containing credentials are not supported")
    _ = parts.port  # Validate port before making a request.
    return urlunsplit((parts.scheme.lower(), parts.netloc, parts.path or "/", parts.query, ""))


def origin(url: str) -> str:
    p = urlsplit(url)
    return urlunsplit((p.scheme, p.netloc, "", "", ""))


def origin_key(url: str) -> tuple[str, str, int]:
    p = urlsplit(url)
    return p.scheme, p.hostname.lower(), p.port if p.port is not None else (443 if p.scheme == "https" else 80)


def authorized_origin(value: str) -> str:
    url = normalize_url(value)
    p = urlsplit(url)
    if p.path != "/" or p.query or urlsplit(value).fragment:
        raise ValueError("Authorized origins must contain only scheme, hostname and optional port")
    return origin(url)


def in_scope(url: str, authorized: set[tuple[str, str, int]]) -> bool:
    key = origin_key(url)
    # Only the ordinary HTTP default port may upgrade to default-port HTTPS.
    return key in authorized or (key[0] == "https" and key[2] == 443 and ("http", key[1], 80) in authorized)


@dataclass
class Response:
    status: int
    headers: dict[str, str]
    body: str
    truncated: bool


@dataclass
class Result:
    input_url: str
    final_url: str = ""
    status: str = "error"
    fetch_status: str = "blocked"
    http_status: int | None = None
    wordpress_version: str = ""
    php_version: str = ""
    server: str = ""
    powered_by: str = ""
    title: str = ""
    evidence: list[str] = field(default_factory=list)
    asset_version_hints: list[str] = field(default_factory=list)
    notes: list[str] = field(default_factory=list)
    checked_at: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat())
    php_version_candidates: list[str] = field(default_factory=list)
    php_evidence: list[str] = field(default_factory=list)
    php2shell_status: str = "unknown"
    php2shell_version_status: str = "version_not_disclosed"
    php2shell_fixed_version: str = ""
    php2shell_exploitability: str = "not_tested"
    php2shell_cves: list[str] = field(default_factory=lambda: list(PHP2SHELL_CVES))
    php2shell_rules_as_of: str = PHP2SHELL_RULES_DATE
    php2shell_sources: list[str] = field(default_factory=lambda: list(PHP2SHELL_SOURCES))
    php2shell_notes: list[str] = field(default_factory=list)
    php2shell_recommendation: str = ""


class PageParser(HTMLParser):
    def __init__(self):
        super().__init__(convert_charrefs=True)
        self.generators: list[str] = []
        self.assets: list[str] = []
        self.api_links: list[str] = []
        self.title_parts: list[str] = []
        self.in_title = False

    def handle_starttag(self, tag, attrs):
        a = dict(attrs)
        if tag == "meta" and (a.get("name") or "").lower() == "generator":
            self.generators.append(a.get("content") or "")
        if tag == "link" and "https://api.w.org/" in (a.get("rel") or "").split():
            self.api_links.append(a.get("href") or "")
        if tag in {"script", "img", "source", "link"}:
            for key in ("src", "href"):
                if a.get(key):
                    self.assets.append(a[key])
        if tag == "title":
            self.in_title = True

    def handle_endtag(self, tag):
        if tag == "title":
            self.in_title = False

    def handle_data(self, data):
        if self.in_title:
            self.title_parts.append(data)


def _inspect_page(response: Response, result: Result) -> None:
    result.http_status = response.status
    result.fetch_status = "ok" if 200 <= response.status < 300 else "blocked"
    result.server = response.headers.get("server", "")
    result.powered_by = response.headers.get("x-powered-by", "")
    candidates = set()
    for header, value in (("X-Powered-By", result.powered_by), ("Server", result.server)):
        if re.search(r"\bPHP\b", value, re.I):
            result.php_evidence.append(f"{header}: {value}")
        candidates.update(re.findall(r"\bPHP/([0-9][A-Za-z0-9.+_~:-]*)(?=$|[\s,;()])", value, re.I))
    result.php_version_candidates = sorted(candidates)
    result.php_version = next(iter(candidates)) if len(candidates) == 1 else ""
    if len(candidates) > 1:
        result.notes.append("Conflicting PHP versions in public headers; patch status is unknown.")
    if response.status in {401, 403, 407, 429, 451}:
        result.status = "blocked"
        result.notes.append(f"HTTP {response.status}; access restriction or rate limit. No retry.")
        return
    if not 200 <= response.status < 300:
        result.status = "error"
        result.notes.append(f"HTTP {response.status}; detection skipped.")
        return
    if response.headers.get("content-encoding", "identity").lower() not in {"", "identity"}:
        result.status = "uncertain"
        result.notes.append("Server sent an unsupported content encoding despite requesting identity.")
        return
    content_type = response.headers.get("content-type", "").lower()
    if content_type and not any(t in content_type for t in ("text/html", "application/xhtml+xml")):
        result.status = "uncertain"
        result.notes.append(f"Expected HTML; received {content_type}.")
        return
    page = PageParser()
    page.feed(response.body)
    result.title = " ".join("".join(page.title_parts).split())[:500]
    versions = set()
    for generator in page.generators:
        match = re.match(r"^WordPress(?:\s+([0-9]+(?:\.[0-9]+)*(?:[-+._a-zA-Z0-9]*)?))?(?:\s|$)", generator, re.I)
        if match:
            result.evidence.append(f"HTML generator: {generator[:200]}")
            if match.group(1):
                versions.add(match.group(1))
    if len(versions) == 1:
        result.wordpress_version = next(iter(versions))
    elif versions:
        result.notes.append("Conflicting WordPress generator versions: " + ", ".join(sorted(versions)))
    if page.api_links or re.search(r'\brel\s*=\s*["\']?https://api\.w\.org/', response.headers.get("link", ""), re.I):
        result.evidence.append("WordPress REST API discovery relation (https://api.w.org/)")
    strong = bool(result.evidence)
    for asset in sorted(set(page.assets)):
        p = urlsplit(urljoin(result.final_url, asset))
        if re.search(r"/(?:wp-content|wp-includes)/", p.path, re.I):
            result.evidence.append(f"WordPress-style asset: {asset[:300]}")
            # An asset's ver parameter can be a plugin/theme version or cache key.
            for version in parse_qs(p.query).get("ver", []):
                result.asset_version_hints.append(f"{version[:80]} ({p.path[:200]})")
    result.status = "wordpress" if strong else "likely_wordpress" if result.evidence else "no_evidence"
    if response.truncated:
        result.notes.append("HTML inspection limited to the first 2 MiB.")
        if result.status == "no_evidence":
            result.status = "uncertain"
    if result.status == "no_evidence":
        result.notes.append("No public WordPress markers found; this does not establish that WordPress is absent.")
    if result.php_version or result.wordpress_version:
        result.notes.append("Versions are server/page disclosures, not independently verified runtime versions.")


def inspect_page(response: Response, result: Result) -> None:
    try:
        _inspect_page(response, result)
    finally:
        annotate_php2shell(result)


class Checker:
    def __init__(self, *, timeout=15.0, delay=1.0, contact="", max_crawl_delay=60.0, authorized_origins=(), target_timeout=None):
        self.timeout = timeout
        self.delay = delay
        self.max_crawl_delay = max_crawl_delay
        if target_timeout is not None and (not math.isfinite(target_timeout) or target_timeout <= 0):
            raise ValueError("target_timeout must be positive and finite")
        self.target_timeout = target_timeout
        self._deadline = None
        self.authorized_origins = {origin_key(authorized_origin(value)) for value in authorized_origins}
        self.user_agent = f"{BOT}/1.0" + (f" (+{contact})" if contact else "")
        self.opener = build_opener(NoRedirect())
        self.last_request: dict[str, float] = {}
        self.robots: dict[str, tuple[RobotsPolicy | None, float, str]] = {}

    def request(self, url: str, limit: int, delay: float | None = None) -> Response:
        own_deadline = self._deadline is None and self.target_timeout is not None
        if own_deadline:
            self._deadline = time.monotonic() + self.target_timeout
        try:
            return self._request(url, limit, delay)
        finally:
            if own_deadline:
                self._deadline = None

    def _remaining(self):
        if self._deadline is None:
            return None
        remaining = self._deadline - time.monotonic()
        if remaining <= 0:
            raise TargetDeadline("Total target deadline exceeded; assessment incomplete.")
        return remaining

    def _request(self, url, limit, delay):
        site = origin(url)
        wait = max(self.delay, delay or 0) - (time.monotonic() - self.last_request.get(site, -math.inf))
        remaining = self._remaining()
        if wait > 0:
            if remaining is not None and wait >= remaining:
                raise TargetDeadline("Total target deadline cannot accommodate required pacing; assessment incomplete.")
            time.sleep(wait)
        self.last_request[site] = time.monotonic()
        remaining = self._remaining()
        if remaining is None:
            return self._fetch(url, limit)
        # A disposable process bounds even resolver calls and trickling headers/body.
        # One child is active at a time; parent retains robots policy and pacing.
        command = [sys.executable, "-I", "-c", "import runpy,sys; runpy.run_path(sys.argv[1])['_request_worker']()", str(Path(__file__).resolve())]
        child = subprocess.Popen(command, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                                 creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0))
        try:
            payload = json.dumps({"url": url, "limit": limit, "timeout": min(self.timeout, remaining), "user_agent": self.user_agent})
            output, _ = child.communicate(payload.encode("utf-8"), timeout=self._remaining())
        except subprocess.TimeoutExpired:
            raise TargetDeadline("Total target deadline exceeded during HTTP acquisition; assessment incomplete.") from None
        finally:
            if child.poll() is None:
                child.kill()
            # Reap the only child before another target can start. No resolver thread survives.
            try:
                child.wait(timeout=2)
            except subprocess.TimeoutExpired:
                # An OS-level cancellation failure must abort, never launch another worker.
                raise RuntimeError("Unable to reap HTTP worker after cancellation; batch aborted") from None
            for pipe in (child.stdin, child.stdout, child.stderr):
                if pipe is not None:
                    pipe.close()
            self.last_request[site] = time.monotonic()
        self._remaining()
        if child.returncode != 0:
            raise OSError("HTTP acquisition worker failed")
        message = json.loads(output)
        if "error" in message:
            kind = message.get("kind")
            error = message["error"]
            if kind == "timeout":
                raise TimeoutError(error)
            if kind == "dns":
                raise socket.gaierror(error)
            raise OSError(error)
        return Response(**message["response"])

    def _fetch(self, url, limit):
        req = Request(url, headers={"User-Agent": self.user_agent, "Accept": "text/html,text/plain;q=0.9,*/*;q=0.1", "Accept-Encoding": "identity"})
        try:
            stream = self.opener.open(req, timeout=self.timeout)
        except HTTPError as exc:
            stream = exc  # Preserve status/headers; never automatically retry.
        with stream:
            raw = stream.read(limit + 1)
            charset = stream.headers.get_content_charset() or "utf-8"
            try:
                body = raw[:limit].decode(charset, errors="replace")
            except LookupError:
                body = raw[:limit].decode("utf-8", errors="replace")
            headers = {k.lower(): ", ".join(stream.headers.get_all(k)) for k in stream.headers.keys()}
            return Response(stream.code, headers, body, len(raw) > limit)

    def policy(self, url: str, authorized=None) -> float:
        site = origin(url)
        if authorized is None:
            authorized = self.authorized_origins | {origin_key(url)}
        if site not in self.robots:
            robot_url = site + "/robots.txt"
            response = None
            seen = set()
            for _ in range(6):
                if robot_url in seen:
                    raise CheckStopped("robots.txt redirect loop; skipping site.")
                seen.add(robot_url)
                response = self.request(robot_url, MAX_ROBOTS)
                if response.status not in REDIRECTS:
                    break
                location = response.headers.get("location")
                if not location:
                    raise CheckStopped("robots.txt redirect without Location; skipping site.")
                target = normalize_url(urljoin(robot_url, location))
                if not in_scope(target, authorized):
                    raise OutOfScope("robots.txt redirects to another origin outside authorized scope; skipping site.")
                if urlsplit(robot_url).scheme == "https" and urlsplit(target).scheme == "http":
                    raise CheckStopped("HTTPS-to-HTTP robots.txt redirect; stopped.")
                if origin_key(target) != origin_key(url) and urlsplit(target).path != "/robots.txt":
                    raise CheckStopped("Cross-origin robots.txt redirect must retain /robots.txt; skipping site.")
                robot_url = target
            assert response is not None
            if response.status in {404, 410}:
                self.robots[site] = (None, self.delay, "robots.txt not published")
            elif response.status == 200 and not response.truncated:
                if response.headers.get("content-encoding", "identity").lower() not in {"identity", ""}:
                    raise CheckStopped("Cannot decode robots.txt; skipping site.")
                if re.search(r"<(?:!doctype\s+html|html|head|body)\b", response.body, re.I):
                    raise CheckStopped("robots.txt returned HTML; policy cannot be verified, skipping site.")
                parser = RobotsPolicy(response.body, self.delay)
                self.robots[site] = (parser, parser.delay, "robots.txt checked")
            else:
                raise CheckStopped(f"Cannot verify robots.txt (HTTP {response.status}, truncated={response.truncated}); skipping site.")
        parser, delay, _ = self.robots[site]
        if parser is not None and not parser.can_fetch(url):
            raise CheckStopped("robots.txt disallows this URL for this checker.")
        if delay > self.max_crawl_delay:
            raise CheckStopped(f"robots.txt requires {delay:g}-second request spacing; exceeds the {self.max_crawl_delay:g}-second batch limit. Page not fetched.")
        return delay

    def check(self, value: str) -> Result:
        result = Result(input_url=value)
        self._deadline = time.monotonic() + self.target_timeout if self.target_timeout is not None else None
        try:
            url = normalize_url(value)
            authorized = self.authorized_origins | {origin_key(url)}
            seen = set()
            for _ in range(6):
                self._remaining()
                result.final_url = url
                if url in seen:
                    raise CheckStopped("Page redirect loop; stopped.")
                seen.add(url)
                delay = self.policy(url, authorized)
                response = self.request(url, MAX_BODY, delay)
                result.http_status = response.status
                if response.status not in REDIRECTS:
                    result.notes.append(self.robots[origin(url)][2])
                    inspect_page(response, result)
                    self._remaining()
                    return result
                location = response.headers.get("location")
                if not location:
                    raise CheckStopped("Page redirect without Location; stopped.")
                target = normalize_url(urljoin(url, location))
                if urlsplit(url).scheme == "https" and urlsplit(target).scheme == "http":
                    raise CheckStopped("HTTPS-to-HTTP redirect; stopped.")
                if not in_scope(target, authorized):
                    raise OutOfScope("Page redirect outside authorized origin scope; stopped before requesting destination.")
                url = target
            raise CheckStopped("Too many page redirects; stopped.")
        except CheckStopped as exc:
            result.status = "blocked"
            if isinstance(exc, TargetDeadline):
                result.status = "error"
                result.fetch_status = "deadline_exceeded"
            if isinstance(exc, OutOfScope):
                result.fetch_status = "out_of_scope"
            result.notes.append(str(exc))
        except (ValueError, URLError, OSError, HTTPException, UnicodeError) as exc:
            result.status = "error"
            cause = exc.reason if isinstance(exc, URLError) else exc
            if isinstance(cause, (TimeoutError, socket.timeout)):
                result.fetch_status = "timeout"
            elif isinstance(cause, socket.gaierror):
                result.fetch_status = "dns_error"
            result.notes.append(str(exc))
        finally:
            self._deadline = None
            if result.fetch_status == "deadline_exceeded":
                # Evidence parsed just as the budget expired is not a completed assessment.
                for name, value in assess_php2shell().items():
                    setattr(result, name, value)
                result.php2shell_notes.append("Total target deadline exceeded; assessment incomplete.")
            else:
                annotate_php2shell(result)
        return result


def _request_worker():
    """Private bounded HTTP acquisition; policy decisions stay in the parent."""
    payload = json.loads(sys.stdin.buffer.read().decode("utf-8"))
    checker = Checker(timeout=payload["timeout"], delay=0)
    checker.user_agent = payload["user_agent"]
    try:
        message = {"response": asdict(checker._fetch(payload["url"], payload["limit"]))}
    except (ValueError, URLError, OSError, HTTPException, UnicodeError) as exc:
        cause = exc.reason if isinstance(exc, URLError) else exc
        kind = "timeout" if isinstance(cause, (TimeoutError, socket.timeout)) else "dns" if isinstance(cause, socket.gaierror) else "error"
        message = {"error": str(exc)[:4096], "kind": kind}
    sys.stdout.buffer.write(json.dumps(message, ensure_ascii=False).encode("utf-8"))


def positive_float(value):
    number = float(value)
    if not math.isfinite(number) or number <= 0:
        raise argparse.ArgumentTypeError("must be a finite number greater than zero")
    return number


def csv_safe(value):
    if isinstance(value, list):
        value = " | ".join(value)
    # Prevent spreadsheet formulas from untrusted page titles / response fields.
    if isinstance(value, str) and value.lstrip().startswith(("=", "+", "-", "@")):
        return "'" + value
    return value


DEFAULT_LOGO = Path(__file__).resolve().parent / "assets" / "logo.png"
STATUS_ORDER = {"potentially_affected": 0, "unknown": 1, "fixed_version_reported": 2}
STATUS_LABELS = {
    "potentially_affected": "Potentially affected",
    "fixed_version_reported": "Fixed version reported",
    "unknown": "Unknown",
}
REPORT_CSS = """
:root{--bg:#F8FAFC;--surface:#FFFFFF;--surface-2:#EEF2FF;--ink:#0F172A;--ink-2:#1E293B;--ink-3:#64748B;--line:#E2E8F0;
--accent:#1E3A8A;--accent-text:#1E3A8A;--on-accent:#FFFFFF;--bad:#B42318;--bad-bg:#FEE4E2;--ok:#067647;--ok-bg:#DCFAE6;--unk:#475467;--unk-bg:#EAECF0}
@media (prefers-color-scheme:dark){:root{--bg:#0F172A;--surface:#1E293B;--surface-2:#273449;--ink:#F1F5F9;--ink-2:#CBD5E1;--ink-3:#94A3B8;
--line:#334155;--accent-text:#93C5FD;--bad:#FDA29B;--bad-bg:#55160C;--ok:#75E0A7;--ok-bg:#053321;--unk:#D0D5DD;--unk-bg:#344054}}
*{box-sizing:border-box}body{margin:0;background:var(--bg);color:var(--ink);font:15px/1.5 Arial,system-ui,sans-serif}
main{max-width:1080px;margin:0 auto;padding:24px 16px 48px}
header{display:flex;gap:20px;align-items:center;padding:20px 0;border-bottom:3px solid var(--accent);margin-bottom:20px}
header img{height:64px;width:auto;border-radius:8px;background:#F8FAFC;padding:4px}
.eyebrow{font-size:12px;letter-spacing:.12em;font-weight:700;color:var(--accent-text);text-transform:uppercase}
h1{margin:2px 0;font-size:26px}.sub{color:var(--ink-3);margin:0}h2{margin:32px 0 10px;font-size:18px;color:var(--ink-2)}
.tiles{display:grid;grid-template-columns:repeat(auto-fit,minmax(160px,1fr));gap:12px}
.tile{background:var(--surface);border:1px solid var(--line);border-radius:10px;padding:14px}
.tile b{display:block;font-size:28px;color:var(--accent-text)}.tile span{color:var(--ink-3);font-size:13px}
.tile.bad b{color:var(--bad)}.tile.ok b{color:var(--ok)}
.wrap{overflow-x:auto}table{width:100%;border-collapse:collapse;background:var(--surface);border:1px solid var(--line)}
table.narrow{max-width:360px}
th{background:var(--accent);color:var(--on-accent);text-align:left;padding:8px 10px;font-size:13px;white-space:nowrap}
td{padding:8px 10px;border-top:1px solid var(--line);vertical-align:top;font-size:14px}tr:nth-child(even) td{background:var(--surface-2)}
td.site{overflow-wrap:anywhere;min-width:200px}td.notes{color:var(--ink-3);font-size:13px;min-width:240px}
code{font:13px Consolas,monospace}
.pill{display:inline-block;padding:2px 10px;border-radius:999px;font-size:12px;font-weight:700;white-space:nowrap}
.pill.potentially_affected{color:var(--bad);background:var(--bad-bg)}.pill.fixed_version_reported{color:var(--ok);background:var(--ok-bg)}
.pill.unknown{color:var(--unk);background:var(--unk-bg)}
.callout{background:var(--surface);border:1px solid var(--line);border-left:4px solid var(--accent);border-radius:8px;padding:12px 16px;color:var(--ink-2)}
footer{margin-top:40px;padding-top:12px;border-top:1px solid var(--line);color:var(--ink-3);font-size:12px}
a{color:var(--accent-text)}
@media print{body{background:#fff}tr,.tile{break-inside:avoid}}
"""


def logo_data_uri(path) -> str:
    """Inline a local logo so the report stays a single offline file; a missing logo is omitted."""
    try:
        data = Path(path).read_bytes()
    except OSError:
        return ""
    mime = mimetypes.guess_type(str(path))[0] or "image/png"
    return f"data:{mime};base64,{base64.b64encode(data).decode('ascii')}"


def render_html(rows, *, logo=DEFAULT_LOGO, brand="CompleteTech LLC", generated_at=None, rules_as_of=PHP2SHELL_RULES_DATE) -> str:
    """Render result dictionaries as a self-contained, script-free HTML report."""
    def esc(value):
        return html.escape("" if value is None else str(value))

    generated_at = generated_at or datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC")
    rows = sorted(rows, key=lambda row: STATUS_ORDER.get(row.get("php2shell_status"), 1))
    counts = {key: sum(row.get("php2shell_status") == key for row in rows) for key in STATUS_ORDER}
    tiles = (
        ("", len(rows), "Websites checked"),
        ("bad", counts["potentially_affected"], "Potentially affected"),
        ("ok", counts["fixed_version_reported"], "Fixed version reported"),
        ("", counts["unknown"], "Unknown / not disclosed"),
        ("", sum(row.get("status") in ("wordpress", "likely_wordpress") for row in rows), "WordPress detected"),
    )
    body = []
    for row in rows:
        status = row.get("php2shell_status") or "unknown"
        notes = row.get("php2shell_notes") or []
        body.append(
            "<tr>"
            f"<td class=site>{esc(row.get('final_url') or row.get('input_url'))}<br>Observed: {esc(row.get('checked_at') or 'unknown')}</td>"
            f"<td><span class='pill {esc(status)}'>{esc(STATUS_LABELS.get(status, status))}</span></td>"
            f"<td><code>{esc(row.get('php_version') or '—')}</code></td>"
            f"<td><code>{esc(row.get('php2shell_fixed_version') or '—')}</code></td>"
            f"<td>{esc(row.get('status'))}</td>"
            f"<td>{esc(row.get('http_status') or '—')}</td>"
            f"<td class=notes>{esc(notes[-1] if notes else row.get('php2shell_recommendation'))}</td>"
            "</tr>"
        )
    fixes = "".join(f"<tr><td>{major}.{minor}</td><td><code>{major}.{minor}.{patch}</code></td></tr>" for (major, minor), patch in PHP2SHELL_FIXES.items())
    logo_uri = logo_data_uri(logo) if logo else ""
    logo_html = f"<img src='{logo_uri}' alt='{esc(brand)} logo'>" if logo_uri else ""
    tile_html = "".join(f"<div class='tile {cls}'><b>{value}</b><span>{esc(label)}</span></div>" for cls, value, label in tiles)
    sources = " · ".join(f"<a href='{esc(url)}'>{esc(url.rsplit('/', 1)[-1])}</a>" for url in PHP2SHELL_SOURCES)
    empty = "<tr><td colspan=7>No results.</td></tr>"
    return f"""<!doctype html>
<html lang="en"><head><meta charset="utf-8"><meta name="viewport" content="width=device-width,initial-scale=1">
<meta http-equiv="Content-Security-Policy" content="default-src 'none'; img-src data:; style-src 'unsafe-inline'">
<title>PHP2Shell Exposure Report</title><style>{REPORT_CSS}</style></head>
<body><main>
<header>{logo_html}<div><div class="eyebrow">{esc(brand)}</div><h1>PHP2Shell Exposure Report</h1>
<p class="sub">Current reference CVEs: {' &amp; '.join(PHP2SHELL_CVES)} · acquisition generated {esc(generated_at)} · acquisition rules as of {esc(rules_as_of)}</p></div></header>
<div class="tiles">{tile_html}</div>
<h2>Results</h2>
<div class="wrap"><table><thead><tr><th>Website</th><th>PHP2Shell</th><th>PHP reported</th><th>Upstream fix</th><th>Detection</th><th>HTTP</th><th>Notes</th></tr></thead>
<tbody>{''.join(body) or empty}</tbody></table></div>
<h2>Current reference fix thresholds</h2>
<p class="callout">Current reference rules as of {esc(PHP2SHELL_RULES_DATE)}. This table and the current source links are guidance from the rendering checker; they do not reclassify preserved acquisition results.</p>
<div class="wrap"><table class="narrow"><thead><tr><th>Branch</th><th>Fixed in</th></tr></thead><tbody>{fixes}</tbody></table></div>
<h2>How to read this report</h2>
<p class="callout"><b>Potential exposure is not confirmed exploitability.</b> Results compare PHP versions that servers advertise
in public headers with upstream fix releases. No exploit payloads were sent. Distributions can backport fixes without changing the
version number, and hidden versions are reported as unknown, not safe. Administrators should verify the installed package and runtime.</p>
<footer>{esc(brand)} · Generated by php2shell-check · Current reference sources: {sources}</footer>
</main></body></html>
"""



MAX_CHECKPOINT_BYTES = 64 * 1024 * 1024
MAX_CHECKPOINT_ROWS = 100000


def checkpoint_json(report):
    text = json.dumps(report, ensure_ascii=False)
    if len(text.encode("utf-8")) > MAX_CHECKPOINT_BYTES:
        raise ValueError("Checkpoint exceeds recovery byte limit; last committed checkpoint retained")
    return text


def publish_report(path, text, *, replace=False):
    """Publish one fully fsynced artifact; no-clobber is enforced at publication."""
    descriptor, temporary = tempfile.mkstemp(prefix=".php2shell-", dir=path.parent)
    temporary = Path(temporary)
    try:
        with os.fdopen(descriptor, "w", encoding="utf-8", newline="") as stream:
            stream.write(text)
            stream.flush()
            os.fsync(stream.fileno())
        if replace:
            os.replace(temporary, path)
        else:
            os.link(temporary, path)  # Fail closed when atomic no-clobber is unavailable.
    finally:
        temporary.unlink(missing_ok=True)


def report_provenance():
    # Resolve the executing file only; evidence/checkpoint symlink policy is unchanged.
    source = Path(__file__).resolve(strict=True)
    descriptor = os.open(source, os.O_RDONLY | getattr(os, "O_BINARY", 0) | getattr(os, "O_NONBLOCK", 0) | getattr(os, "O_NOFOLLOW", 0))
    try:
        if not stat.S_ISREG(os.fstat(descriptor).st_mode):
            raise ValueError("Checker identity requires a regular executable file")
        with os.fdopen(descriptor, "rb", closefd=False) as stream:
            data = stream.read(16 * 1024 * 1024 + 1)
    finally:
        os.close(descriptor)
    if len(data) > 16 * 1024 * 1024:
        raise ValueError("Checker executable exceeds identity size limit")
    rules = {"as_of": PHP2SHELL_RULES_DATE, "cves": PHP2SHELL_CVES,
             "fixes": sorted((list(key), value) for key, value in PHP2SHELL_FIXES.items()),
             "sources": PHP2SHELL_SOURCES}
    return {"tool": TOOL_NAME, "tool_version": TOOL_VERSION,
            "executable_sha256": hashlib.sha256(data).hexdigest(),
            "rules_sha256": hashlib.sha256(json.dumps(rules, sort_keys=True, separators=(",", ":")).encode("utf-8")).hexdigest(),
            "rules_as_of": PHP2SHELL_RULES_DATE,
            "generated_at": datetime.now(timezone.utc).isoformat()}


def validate_provenance(value):
    if not isinstance(value, dict) or set(value) != {"tool", "tool_version", "executable_sha256", "rules_sha256", "rules_as_of", "generated_at"}:
        raise ValueError("Invalid checkpoint provenance fields")
    if value["tool"] != TOOL_NAME or not isinstance(value["tool_version"], str) or not re.fullmatch(r"[0-9]+\.[0-9]+\.[0-9]+", value["tool_version"]):
        raise ValueError("Invalid checkpoint tool identity")
    for key in ("executable_sha256", "rules_sha256"):
        if not isinstance(value[key], str) or not re.fullmatch(r"[0-9a-f]{64}", value[key]):
            raise ValueError("Invalid checkpoint digest")
    for key in ("rules_as_of", "generated_at"):
        if not isinstance(value[key], str):
            raise ValueError("Invalid checkpoint provenance date")
    if not re.fullmatch(r"[0-9]{4}-[0-9]{2}-[0-9]{2}", value["rules_as_of"]):
        raise ValueError("Invalid checkpoint rules date")
    try:
        stamp = datetime.fromisoformat(value["generated_at"])
    except ValueError as exc:
        raise ValueError("Invalid checkpoint generation time") from exc
    if stamp.utcoffset() is None or stamp.utcoffset().total_seconds() != 0:
        raise ValueError("Checkpoint generation time must be UTC")


def run_report(rows, total, *, complete=False, provenance=None):
    return {"journal_schema": "php2shell-run-v1", "provenance": report_provenance() if provenance is None else provenance, "complete": complete,
            "completed_targets": len(rows), "pending_targets": total - len(rows),
            "total_targets": total, "results": list(rows)}


def load_checkpoint(path):
    if path.is_symlink():
        raise ValueError("Recovery refuses symbolic-link checkpoints")
    flags = os.O_RDONLY | getattr(os, "O_BINARY", 0) | getattr(os, "O_NONBLOCK", 0) | getattr(os, "O_NOFOLLOW", 0)
    descriptor = os.open(path, flags)
    try:
        if not stat.S_ISREG(os.fstat(descriptor).st_mode):
            raise ValueError("Recovery requires a regular checkpoint file")
        with os.fdopen(descriptor, "rb", closefd=False) as stream:
            raw = stream.read(MAX_CHECKPOINT_BYTES + 1)
    finally:
        os.close(descriptor)
    if len(raw) > MAX_CHECKPOINT_BYTES:
        raise ValueError("Checkpoint exceeds 64 MiB recovery limit")
    report = json.loads(raw)
    if not isinstance(report, dict) or set(report) not in ({"journal_schema", "complete", "completed_targets", "pending_targets", "total_targets", "results"}, {"journal_schema", "provenance", "complete", "completed_targets", "pending_targets", "total_targets", "results"}):
        raise ValueError("Invalid checkpoint fields")
    if "provenance" in report:
        validate_provenance(report["provenance"])
    rows = report["results"]
    if (report["journal_schema"] != "php2shell-run-v1" or type(report["complete"]) is not bool
            or not isinstance(rows, list) or len(rows) > MAX_CHECKPOINT_ROWS
            or any(type(report[k]) is not int or report[k] < 0 for k in ("completed_targets", "pending_targets", "total_targets"))
            or report["completed_targets"] != len(rows)
            or report["completed_targets"] + report["pending_targets"] != report["total_targets"]
            or (report["complete"] and report["pending_targets"])):
        raise ValueError("Invalid checkpoint completeness")
    example = asdict(Result(input_url=""))
    for row in rows:
        if not isinstance(row, dict) or set(row) != set(example):
            raise ValueError("Invalid checkpoint result fields")
        if row["php2shell_exploitability"] != "not_tested":
            raise ValueError("Checkpoint cannot attest exploitability")
        for key, value in row.items():
            default = example[key]
            if isinstance(default, list):
                valid = isinstance(value, list) and all(isinstance(item, str) for item in value)
            elif key == "http_status":
                valid = value is None or (type(value) is int and 100 <= value <= 999)
            else:
                valid = type(value) is type(default)
            if not valid:
                raise ValueError("Invalid checkpoint result value")
    return report


def format_report(report, args):
    metadata = {key: report[key] for key in ("complete", "completed_targets", "pending_targets", "total_targets")}
    provenance = report.get("provenance")
    # Legacy journals have no acquisition identity; never attribute them to this executable.
    identity = provenance or {"tool": "unknown", "tool_version": "unknown", "executable_sha256": "unknown",
                              "rules_sha256": "unknown", "rules_as_of": "unknown", "generated_at": "unknown"}
    if args.format == "json":
        return json.dumps({**metadata, "provenance": provenance, "results": report["results"]}, ensure_ascii=False, indent=2) + "\n"
    if args.format == "html":
        rendered = render_html(report["results"], logo=args.logo, brand=args.brand, generated_at=identity["generated_at"], rules_as_of=identity["rules_as_of"])
        state = "complete" if report["complete"] else "incomplete / recovered"
        banner = f"<p class=callout>Run {state}: {report['completed_targets']} completed; {report['pending_targets']} pending; {report['total_targets']} total targets.</p>"
        details = "<p class=callout>Acquisition provenance: " + "; ".join(html.escape(key + "=" + str(value)) for key, value in identity.items()) + "</p>"
        return rendered.replace('<h2>Results</h2>', banner + details + '<h2>Results</h2>', 1)
    stream = io.StringIO(newline="")
    fields = list(Result.__dataclass_fields__) + ["record_type", *metadata, *identity]
    writer = csv.DictWriter(stream, fieldnames=fields)
    writer.writeheader()
    for row in report["results"]:
        writer.writerow({key: csv_safe(value) for key, value in {**row, "record_type": "result", **metadata, **identity}.items()})
    if not report["results"]:
        writer.writerow({key: csv_safe(value) for key, value in {"record_type": "summary", **metadata, **identity}.items()})
    return stream.getvalue()


def main(argv=None) -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("urls", nargs="*", help="Website URLs or bare domains (defaults to HTTPS)")
    parser.add_argument("--input", type=Path, help="UTF-8 text file: one URL per line; # comments allowed")
    parser.add_argument("--output", type=Path, help="Atomically save results; existing paths refused unless --force")
    parser.add_argument("--force", action="store_true", help="Explicitly replace an existing final output only after successful completion")
    parser.add_argument("--recover", type=Path, help="Render a persisted .partial.json checkpoint without making requests")
    parser.add_argument("--format", choices=("csv", "json", "html"), default="csv", help="csv, json, or a branded html report")
    parser.add_argument("--delay", type=positive_float, default=1.0, help="Minimum seconds between requests to an origin (default: 1)")
    parser.add_argument("--timeout", type=positive_float, default=15.0, help="Socket timeout per request (default: 15 seconds)")
    parser.add_argument("--target-timeout", type=positive_float, default=120.0, help="Total monotonic seconds per target, including DNS, HTTP, redirects and pacing (default: 120)")
    parser.add_argument("--authorized-origin", action="append", default=[], type=authorized_origin, help="Additional authorized redirect origin (scheme, host and optional port); repeatable")
    parser.add_argument("--contact", default="", help="Public contact URL/email to include in the honest User-Agent")
    parser.add_argument("--logo", type=Path, default=DEFAULT_LOGO, help="Logo image for the html report (default: assets/logo.png)")
    parser.add_argument("--brand", default="CompleteTech LLC", help="Organization name shown on the html report")
    args = parser.parse_args(argv)
    if any(ord(c) < 32 or ord(c) == 127 for c in args.contact):
        parser.error("--contact must not contain control characters")
    urls = list(args.urls)
    if args.input:
        try:
            urls.extend(line.strip() for line in args.input.read_text(encoding="utf-8-sig").splitlines() if line.strip() and not line.lstrip().startswith("#"))
        except OSError as exc:
            parser.error(str(exc))
    urls = list(dict.fromkeys(urls))
    if not urls and not args.recover:
        parser.error("provide website URLs or --input websites.txt")
    if len(urls) > MAX_CHECKPOINT_ROWS:
        parser.error("At most 100,000 targets per run; split inputs explicitly")
    if args.recover and (urls or args.input):
        parser.error("--recover cannot be combined with targets or --input")
    if args.force and not args.output:
        parser.error("--force requires --output")
    if args.input and args.output and args.input.resolve() == args.output.resolve():
        parser.error("--output must differ from --input")
    checkpoint = Path(str(args.output) + ".partial.json") if args.output and not args.recover else None
    if args.output:
        if args.output.is_symlink() or (args.output.exists() and (not args.force or not args.output.is_file())):
            parser.error("Output already exists or is not a regular file; choose a new path or explicitly --force")
        if args.recover and args.output.resolve() == args.recover.resolve():
            parser.error("Recovery output must differ from its checkpoint")
    try:
        if args.recover:
            report = load_checkpoint(args.recover)
        else:
            report = run_report([], len(urls))
            if checkpoint:
                # Exclusive creation protects a previous interrupted run from accidental reuse.
                publish_report(checkpoint, checkpoint_json(report))
            checker = Checker(timeout=args.timeout, delay=args.delay, contact=args.contact, authorized_origins=args.authorized_origin, target_timeout=args.target_timeout)
            try:
                for index, url in enumerate(urls):
                    result = checker.check(url)
                    report = run_report([*report["results"], asdict(result)], len(urls), provenance=report["provenance"])
                    if checkpoint:
                        publish_report(checkpoint, checkpoint_json(report), replace=True)
                    print(f"[{index + 1}/{len(urls)}] {url}: {result.status}", file=sys.stderr)
                report["complete"] = True
            except KeyboardInterrupt:
                if not checkpoint:
                    sys.stdout.write(format_report(report, args))
                    sys.stdout.flush()
                print("Interrupted: run incomplete; completed results remain in the checkpoint." if checkpoint else "Interrupted: output is incomplete.", file=sys.stderr)
                return 130
        rendered = format_report(report, args)
        if args.output:
            publish_report(args.output, rendered, replace=args.force)
        else:
            sys.stdout.write(rendered)
            sys.stdout.flush()
        # Retain the conservative checkpoint, even after success: recovery never invents completion.
        return 0 if report["complete"] else 130
    except (OSError, ValueError, TypeError) as exc:
        print(f"Report failed: {exc}. Existing output and persisted checkpoint are preserved.", file=sys.stderr)
        return 2


if __name__ == "__main__":
    raise SystemExit(main())