1from __future__ import annotations
  2
  3from typing import TYPE_CHECKING
  4
  5import httpx
  6
  7from . import __version__
  8from .audits import (
  9    ContentTypeOptionsAudit,
 10    CookiesAudit,
 11    CORSAudit,
 12    CSPAudit,
 13    FrameOptionsAudit,
 14    HSTSAudit,
 15    RedirectsAudit,
 16    ReferrerPolicyAudit,
 17    StatusCodeAudit,
 18    TLSAudit,
 19)
 20from .metadata import ScanMetadata
 21
 22if TYPE_CHECKING:
 23    from .audits.base import Audit
 24    from .results import ScanResult
 25
 26__all__ = ["Scanner"]
 27
 28
 29class Scanner:
 30    """Main scanner that runs security checks against a URL."""
 31
 32    def __init__(self, url: str, disabled_audits: set[str] | None = None) -> None:
 33        self.url = url
 34        self.disabled_audits = disabled_audits or set()
 35        self.response: httpx.Response | None = None
 36        self.fetch_exception: Exception | None = None
 37
 38        # Initialize all available audits
 39        # Required audits first, then optional ones
 40        self.audits: list[Audit] = [
 41            # Required security audits
 42            StatusCodeAudit(),  # Check status code first (most fundamental)
 43            CSPAudit(),
 44            HSTSAudit(),
 45            TLSAudit(),
 46            RedirectsAudit(),
 47            ContentTypeOptionsAudit(),
 48            FrameOptionsAudit(),
 49            ReferrerPolicyAudit(),
 50            # Optional audits
 51            CookiesAudit(),
 52            CORSAudit(),
 53        ]
 54
 55    def fetch(self) -> httpx.Response:
 56        """Fetch the URL and cache the response."""
 57        if self.response is None:
 58            try:
 59                user_agent = (
 60                    f"plain-scan/{__version__} (+https://plainframework.com/scan)"
 61                )
 62                self.response = httpx.get(
 63                    self.url,
 64                    follow_redirects=True,
 65                    timeout=30,
 66                    headers={"User-Agent": user_agent},
 67                )
 68            except httpx.TransportError as e:
 69                # Store TLS/network exceptions so TLSAudit can report them.
 70                # TransportError covers cert failures, connection refused, and
 71                # connect/read timeouts alike (httpx splits these across
 72                # ConnectError and TimeoutException, both TransportError).
 73                self.fetch_exception = e
 74                raise
 75        return self.response
 76
 77    def scan(self) -> ScanResult:
 78        """Run all security checks and return results."""
 79        from .results import ScanResult
 80
 81        # Try to fetch the URL once
 82        # If this fails with TLS/network errors, we store the exception
 83        # and continue so TLSAudit can report the issue
 84        response = None
 85        try:
 86            response = self.fetch()
 87        except httpx.TransportError:
 88            # Exception is already stored in self.fetch_exception
 89            # Continue with scan so TLSAudit can report it
 90            pass
 91
 92        # Collect metadata about the request
 93        metadata = ScanMetadata.from_response(response)
 94
 95        # Run each audit
 96        scan_result = ScanResult(url=self.url, metadata=metadata)
 97        for audit in self.audits:
 98            # If audit is disabled by user, add to results but mark as disabled
 99            if audit.slug in self.disabled_audits:
100                from .results import AuditResult
101
102                scan_result.audits.append(
103                    AuditResult(
104                        name=audit.name,
105                        detected=False,
106                        required=audit.required,
107                        checks=[],
108                        disabled=True,
109                        description=audit.description,
110                    )
111                )
112            else:
113                # Try to run the audit
114                # If the initial fetch failed and this audit needs the response,
115                # it will fail. TLSAudit handles fetch exceptions specially.
116                try:
117                    audit_result = audit.check(self)
118                    scan_result.audits.append(audit_result)
119                except httpx.TransportError:
120                    # Audit couldn't run due to fetch failure
121                    # Skip non-TLS audits since they need a successful response
122                    if audit.slug != "tls":
123                        from .results import AuditResult, CheckResult
124
125                        scan_result.audits.append(
126                            AuditResult(
127                                name=audit.name,
128                                detected=False,
129                                required=audit.required,
130                                checks=[
131                                    CheckResult(
132                                        name="Connection",
133                                        passed=False,
134                                        message="Could not connect to URL to run audit",
135                                    )
136                                ],
137                                description=audit.description,
138                            )
139                        )
140                    else:
141                        # TLS audit should have handled this - re-raise
142                        raise
143
144        return scan_result