Expose SRI analysis as an importable library API (#18)

README.md

@@ -219,6 +219,30 @@ and short reason codes for each unsafe include. When `--list-sri` is supplied th
 output also enumerates each external script and stylesheet that defines an
 `integrity` attribute.
 
+### Use as a library
+
+The same analysis is available as an importable API from the installed package,
+so other tools (e.g. a findings/dispute workflow) can reuse it instead of
+reimplementing SRI parsing:
+
+```python
+from domain_security_analyzer.sri import scan_url
+
+# Inspect a single page (default); pass crawl=True to follow same-origin links.
+report = scan_url("https://example.com")
+
+for resource in report["unsafe_resources"]:
+    print(resource["resource_url"], resource["reasons"])
+
+if report["compensating_control_detected"]:
+    print("A restrictive Content-Security-Policy is acting as a compensating control")
+```
+
+Each entry in `unsafe_resources` carries the raw `integrity`/`crossorigin`
+values plus machine-readable `reasons`: `missing-integrity`,
+`invalid-integrity-hash`, `mixed-invalid-hashes`, `non-https-resource`, and
+`missing-crossorigin`. For full control, use the `SRIParser` class directly.
+
 ## Documentation
 
 ### **Reference Guides**

domain_security_analyzer/__init__.py

@@ -7,5 +7,13 @@
 
 from .__version__ import __version__
 from .analyzer import DomainAnalyzer, analyze_domains_from_file
+from .sri import SRIParser, UnsafeResource, scan_url
 
-__all__ = ["DomainAnalyzer", "analyze_domains_from_file", "__version__"]
+__all__ = [
+    "DomainAnalyzer",
+    "analyze_domains_from_file",
+    "SRIParser",
+    "UnsafeResource",
+    "scan_url",
+    "__version__",
+]

domain_security_analyzer/sri.py

@@ -0,0 +1,337 @@
+"""Subresource Integrity (SRI) analysis.
+
+Importable API for detecting unsafe Subresource Integrity usage on a site,
+modelled after SecurityScorecard's "Unsafe Implementation of Subresource
+Integrity (SRI)" finding.
+
+Typical use::
+
+    from domain_security_analyzer.sri import scan_url
+
+    report = scan_url("https://example.com")
+    for resource in report["unsafe_resources"]:
+        print(resource["resource_url"], resource["reasons"])
+
+Each unsafe resource carries machine-readable ``reasons`` drawn from:
+
+* ``missing-integrity``     - external resource has no ``integrity`` attribute
+* ``invalid-integrity-hash``- ``integrity`` present but no valid sha256/384/512 hash
+* ``mixed-invalid-hashes``  - some hashes valid, some malformed
+* ``non-https-resource``    - loaded over plain HTTP
+* ``missing-crossorigin``   - cross-origin resource without a ``crossorigin`` attribute
+
+The report also surfaces any ``Content-Security-Policy`` headers and whether a
+restrictive one is present as a compensating control.
+"""
+from __future__ import annotations
+
+import collections
+import re
+from dataclasses import dataclass, field
+from typing import Dict, Iterable, List, Optional, Set, Tuple
+from urllib.parse import urljoin, urlparse
+
+import requests
+from bs4 import BeautifulSoup
+
+__all__ = ["SRIParser", "UnsafeResource", "scan_url", "INTEGRITY_PATTERN"]
+
+# Regular expression to validate integrity hashes (sha256/sha384/sha512)
+INTEGRITY_PATTERN = re.compile(r"^(sha(256|384|512))-[A-Za-z0-9+/=]+$")
+
+# Tokens in script-src that indicate the policy is not restrictive enough to
+# qualify as a compensating control.
+PERMISSIVE_SCRIPT_TOKENS = {
+    "*",
+    "http:",
+    "https:",
+    "data:",
+    "blob:",
+    "filesystem:",
+    "'unsafe-inline'",
+    "'unsafe-eval'",
+}
+
+DEFAULT_USER_AGENT = (
+    "SRI-Parser/1.0 (+https://github.com/CallMarcus/domain-security-analyzer)"
+)
+
+
+@dataclass
+class UnsafeResource:
+    page_url: str
+    resource_url: str
+    tag_type: str
+    integrity: Optional[str]
+    crossorigin: Optional[str]
+    reasons: List[str] = field(default_factory=list)
+
+
+class SRIParser:
+    """Crawl a site and report unsafe Subresource Integrity usage."""
+
+    def __init__(
+        self,
+        base_url: str,
+        max_depth: int = 0,
+        max_pages: int = 1,
+        timeout: int = 10,
+        user_agent: str = DEFAULT_USER_AGENT,
+    ) -> None:
+        parsed = urlparse(base_url)
+        if not parsed.scheme:
+            base_url = f"https://{base_url}"
+            parsed = urlparse(base_url)
+
+        if not parsed.netloc:
+            raise ValueError("A valid domain or URL is required")
+
+        self.base_url = base_url.rstrip('/') or base_url
+        self.base_netloc = parsed.netloc.lower()
+        self.max_depth = max_depth
+        self.max_pages = max_pages
+        self.timeout = timeout
+        self.session = requests.Session()
+        self.session.headers.update({"User-Agent": user_agent})
+
+        self.visited: Set[str] = set()
+        self.to_visit: collections.deque[Tuple[str, int]] = collections.deque([(self.base_url, 0)])
+        self.unsafe_resources: List[UnsafeResource] = []
+        self.csp_policies: List[Tuple[str, str, str]] = []  # (page_url, header_name, policy_value)
+        self.resources_with_integrity: List[Dict[str, object]] = []
+
+    # ------------------------------------------------------------------
+    # Crawling helpers
+    # ------------------------------------------------------------------
+    def _is_same_origin(self, url: str) -> bool:
+        if not url:
+            return False
+        parsed = urlparse(url)
+        if not parsed.scheme:
+            return True  # relative URL -> same origin
+        return parsed.netloc.lower() == self.base_netloc
+
+    def _is_external_resource(self, url: str) -> bool:
+        if not url or not url.startswith(("http://", "https://")):
+            return False
+        parsed = urlparse(url)
+        resource_domain = parsed.netloc.lower().replace("www.", "")
+        base_domain = self.base_netloc.replace("www.", "")
+        return resource_domain != base_domain
+
+    def _fetch(self, url: str) -> Optional[requests.Response]:
+        try:
+            response = self.session.get(url, timeout=self.timeout, allow_redirects=True)
+            if "text/html" not in response.headers.get("content-type", ""):
+                return None
+            return response
+        except requests.RequestException:
+            return None
+
+    def _extract_links(self, html: str, page_url: str) -> Iterable[str]:
+        soup = BeautifulSoup(html, "html.parser")
+        for anchor in soup.find_all("a", href=True):
+            href = urljoin(page_url, anchor.get("href"))
+            if self._is_same_origin(href):
+                yield href.split("#", 1)[0]
+
+    # ------------------------------------------------------------------
+    # SRI evaluation
+    # ------------------------------------------------------------------
+    def _parse_integrity_tokens(self, integrity: str) -> Tuple[List[str], List[str]]:
+        valid_tokens: List[str] = []
+        invalid_tokens: List[str] = []
+        for token in integrity.split():
+            if INTEGRITY_PATTERN.match(token.strip()):
+                valid_tokens.append(token)
+            else:
+                invalid_tokens.append(token)
+        return valid_tokens, invalid_tokens
+
+    def _analyze_resource(self, tag, tag_type: str, page_url: str) -> Optional[UnsafeResource]:
+        src_attr = "src" if tag_type == "script" else "href"
+        resource_url = tag.get(src_attr)
+        if not resource_url:
+            return None
+
+        resource_url = urljoin(page_url, resource_url)
+        if not self._is_external_resource(resource_url):
+            return None
+
+        integrity = tag.get("integrity")
+        crossorigin = tag.get("crossorigin")
+        parsed = urlparse(resource_url)
+        reasons: List[str] = []
+
+        valid_hashes: List[str] = []
+        invalid_hashes: List[str] = []
+
+        if integrity:
+            valid_hashes, invalid_hashes = self._parse_integrity_tokens(integrity)
+            self.resources_with_integrity.append(
+                {
+                    "page_url": page_url,
+                    "resource_url": resource_url,
+                    "tag_type": tag_type,
+                    "integrity": integrity,
+                    "crossorigin": crossorigin,
+                    "valid_hashes": valid_hashes,
+                    "invalid_hashes": invalid_hashes,
+                }
+            )
+        else:
+            reasons.append("missing-integrity")
+
+        if integrity:
+            if not valid_hashes:
+                reasons.append("invalid-integrity-hash")
+            elif invalid_hashes:
+                reasons.append("mixed-invalid-hashes")
+
+        if parsed.scheme != "https":
+            reasons.append("non-https-resource")
+
+        if self._is_cross_origin(parsed) and not crossorigin:
+            reasons.append("missing-crossorigin")
+
+        if reasons:
+            return UnsafeResource(
+                page_url=page_url,
+                resource_url=resource_url,
+                tag_type=tag_type,
+                integrity=integrity,
+                crossorigin=crossorigin,
+                reasons=reasons,
+            )
+        return None
+
+    def _is_cross_origin(self, parsed_url) -> bool:
+        return parsed_url.netloc.lower() != self.base_netloc
+
+    # ------------------------------------------------------------------
+    # CSP evaluation
+    # ------------------------------------------------------------------
+    def _record_csp(self, page_url: str, response: requests.Response) -> None:
+        for header_name in ("Content-Security-Policy", "Content-Security-Policy-Report-Only"):
+            policy = response.headers.get(header_name)
+            if policy:
+                self.csp_policies.append((page_url, header_name, policy))
+
+    def _has_compensating_csp(self) -> bool:
+        for _, _, policy in self.csp_policies:
+            directives = self._parse_csp(policy)
+            script_sources = directives.get("script-src") or directives.get("default-src")
+            if not script_sources:
+                continue
+
+            lower_tokens = [token.lower() for token in script_sources]
+            if any(token in PERMISSIVE_SCRIPT_TOKENS for token in lower_tokens):
+                continue
+
+            has_specific_allowlist = any(
+                token.startswith("'self'")
+                or token.startswith("'nonce-")
+                or token.startswith("'sha")
+                or token.startswith("https://")
+                or token.startswith("http://")
+                or "." in token
+                for token in lower_tokens
+            )
+            if has_specific_allowlist:
+                return True
+        return False
+
+    def _parse_csp(self, policy: str) -> Dict[str, List[str]]:
+        directives: Dict[str, List[str]] = {}
+        for directive in policy.split(";"):
+            directive = directive.strip()
+            if not directive:
+                continue
+            parts = directive.split()
+            if not parts:
+                continue
+            name = parts[0].lower()
+            sources = parts[1:]
+            directives[name] = sources
+        return directives
+
+    # ------------------------------------------------------------------
+    # Public API
+    # ------------------------------------------------------------------
+    def crawl(self) -> Dict[str, object]:
+        pages_crawled = 0
+        while self.to_visit and pages_crawled < self.max_pages:
+            url, depth = self.to_visit.popleft()
+            if url in self.visited or depth > self.max_depth:
+                continue
+
+            response = self._fetch(url)
+            self.visited.add(url)
+            if not response:
+                continue
+
+            pages_crawled += 1
+            html = response.text
+            self._record_csp(url, response)
+
+            soup = BeautifulSoup(html, "html.parser")
+            for script in soup.find_all("script", src=True):
+                unsafe = self._analyze_resource(script, "script", url)
+                if unsafe:
+                    self.unsafe_resources.append(unsafe)
+
+            for link in soup.find_all("link", href=True):
+                rel = link.get("rel") or []
+                if "stylesheet" in [r.lower() for r in rel]:
+                    unsafe = self._analyze_resource(link, "stylesheet", url)
+                    if unsafe:
+                        self.unsafe_resources.append(unsafe)
+
+            if depth < self.max_depth:
+                for href in self._extract_links(html, url):
+                    if href not in self.visited:
+                        self.to_visit.append((href, depth + 1))
+
+        report = {
+            "base_url": self.base_url,
+            "pages_crawled": pages_crawled,
+            "unsafe_resources": [unsafe.__dict__ for unsafe in self.unsafe_resources],
+            "compensating_control_detected": self._has_compensating_csp(),
+            "csp_policies": [
+                {"page_url": page_url, "header": header, "value": policy}
+                for page_url, header, policy in self.csp_policies
+            ],
+            "resources_with_integrity_count": len(self.resources_with_integrity),
+            "resources_with_integrity": self.resources_with_integrity,
+        }
+        return report
+
+
+def scan_url(
+    url: str,
+    crawl: bool = False,
+    max_depth: int = 1,
+    max_pages: int = 25,
+    timeout: int = 10,
+    user_agent: str = DEFAULT_USER_AGENT,
+) -> Dict[str, object]:
+    """Scan a single URL (or crawl a site) for unsafe SRI usage.
+
+    This is the convenience entry point for library consumers such as
+    ``tpcrm-findings-scanner``. By default it inspects only the given page;
+    pass ``crawl=True`` to follow same-origin links up to ``max_depth`` /
+    ``max_pages``.
+
+    Returns the report dict produced by :meth:`SRIParser.crawl`, whose
+    ``unsafe_resources`` entries carry machine-readable ``reasons`` plus the
+    raw ``integrity`` / ``crossorigin`` values, alongside any observed CSP
+    headers and whether a restrictive one acts as a compensating control.
+    """
+    parser = SRIParser(
+        base_url=url,
+        max_depth=max_depth if crawl else 0,
+        max_pages=max_pages if crawl else 1,
+        timeout=timeout,
+        user_agent=user_agent,
+    )
+    return parser.crawl()