From 6407ec68ebee8a9966f8f33fec548c35c2494c14 Mon Sep 17 00:00:00 2001 From: DoRmAmMu1997 Date: Fri, 10 Jul 2026 20:48:49 +0530 Subject: [PATCH 1/2] refactor(IPO-006): one shared SEBI URL canonicalizer + downloader doc riders The listing scraper (sources/sebi.py) and the prospectus downloader (documents/downloader.py) each carried a private _canonical_sebi_url. The copies had drifted: the downloader's gained a malformed-port guard, an optional DNS answer check (anti-rebinding), and a PDF-path restriction the scraper's copy lacked. - backend/ipo/url_canonical.py: single canonical_sebi_url(value, *, base_url, allowed_hosts, error, resolver=None, require_pdf_path=False). Every rejection raises the CALLER's error factory, so each module keeps its own error taxonomy. The raiser-helper shape preserves the original implicit exception chaining (and B904 compliance) exactly. - Both modules keep thin _canonical_sebi_url wrappers with their original signatures, so all call sites and both test suites pass unmodified. - One declared tightening rides along: a malformed port in the scraper now raises SebiSourceError instead of leaking a bare ValueError (rejected either way; now within the module's error taxonomy). Locked by test. - New tests/test_ipo_url_canonical.py (16 tests): canonicalization shape, every rejection path via the injected error type, require_pdf_path on/off, DNS layer (public passes; private/mixed/empty/failing/malformed answers fail closed; resolver=None skips the layer entirely). - Doc riders from the June review, at the code they describe: - %PDF- magic-byte check is header-only BY DESIGN (HTML error pages are the threat; deep validation is the parse stage's job; the content-addressed cache re-verifies the SHA-256 on every later read). - os.replace publish sets no explicit file mode BY DESIGN (public SEBI filings; threat model is tampering -> digest re-verification, not disclosure; data_dir protection is the documented deployment model). Gates: 1,406 passed, coverage 88.17% (floor 87); pre-commit validate, compileall, ruff, mypy (120 files), bandit, pip-audit all clean; IPO docstring/boundary policy guards green. Co-Authored-By: Claude Fable 5 --- backend/ipo/documents/downloader.py | 57 +++++---- backend/ipo/sources/sebi.py | 29 ++--- backend/ipo/url_canonical.py | 101 ++++++++++++++++ tests/test_ipo_url_canonical.py | 174 ++++++++++++++++++++++++++++ 4 files changed, 318 insertions(+), 43 deletions(-) create mode 100644 backend/ipo/url_canonical.py create mode 100644 tests/test_ipo_url_canonical.py diff --git a/backend/ipo/documents/downloader.py b/backend/ipo/documents/downloader.py index 5b60e7a..433da97 100644 --- a/backend/ipo/documents/downloader.py +++ b/backend/ipo/documents/downloader.py @@ -17,7 +17,6 @@ import datetime as dt import enum import hashlib -import ipaddress import os import socket import tempfile @@ -26,12 +25,13 @@ from dataclasses import dataclass from pathlib import Path, PurePosixPath from typing import Any, Never -from urllib.parse import parse_qs, urljoin, urlsplit, urlunsplit +from urllib.parse import parse_qs, urljoin, urlsplit import requests from bs4 import BeautifulSoup from backend.ipo.models import IpoDocumentParseStatus, IpoDocumentRecord +from backend.ipo.url_canonical import canonical_sebi_url ALLOWED_HOSTS = frozenset({"sebi.gov.in", "www.sebi.gov.in"}) ALLOWED_PDF_CONTENT_TYPES = frozenset({"application/pdf", "application/octet-stream"}) @@ -100,35 +100,20 @@ def _canonical_sebi_url( Host allowlisting blocks ordinary SSRF, while resolving the allowlisted host and rejecting non-public answers also catches a poisoned hosts file or DNS response that points SEBI's name at loopback/private infrastructure. - """ - candidate = urljoin(base_url or "", str(value).strip()) - parsed = urlsplit(candidate) - host = (parsed.hostname or "").casefold() - try: - port = parsed.port - except ValueError: - _raise(IpoDocumentDownloadErrorCode.UNSAFE_URL) - if ( - parsed.scheme.casefold() != "https" - or host not in ALLOWED_HOSTS - or parsed.username is not None - or parsed.password is not None - or port not in (None, 443) - ): - _raise(IpoDocumentDownloadErrorCode.UNSAFE_URL) - if require_pdf_path and not parsed.path.startswith("/sebi_data/attachdocs/"): - _raise(IpoDocumentDownloadErrorCode.UNSAFE_URL) - try: - answers = resolver(host, 443, type=socket.SOCK_STREAM) - addresses = {str(answer[4][0]) for answer in answers} - if not addresses or any(not ipaddress.ip_address(address).is_global for address in addresses): - _raise(IpoDocumentDownloadErrorCode.UNSAFE_URL) - except IpoDocumentDownloadError: - raise - except (OSError, TypeError, ValueError, IndexError): - _raise(IpoDocumentDownloadErrorCode.UNSAFE_URL) - return urlunsplit(("https", host, parsed.path or "/", parsed.query, "")) + The implementation is shared with the listing scraper + (``backend/ipo/url_canonical.py``, IPO-006); this wrapper binds the + downloader's secret-safe ``unsafe_url`` error code, its always-on DNS + answer check, and the optional PDF-path restriction. + """ + return canonical_sebi_url( + value, + base_url=base_url or "", + allowed_hosts=ALLOWED_HOSTS, + error=lambda: IpoDocumentDownloadError(IpoDocumentDownloadErrorCode.UNSAFE_URL), + resolver=resolver, + require_pdf_path=require_pdf_path, + ) def _content_type(response: Any) -> str: @@ -383,6 +368,12 @@ def _stream_pdf_to_cache( prefix.extend(chunk[: 5 - len(prefix)]) digest.update(chunk) handle.write(chunk) + # Header-only PDF validation is deliberate (IPO-006 review note): + # the magic-byte check rejects HTML error pages served with a PDF + # content type, while deep structural validation is delegated to + # the parse stage. A truncated or corrupted body is still caught — + # the content-addressed cache stores the SHA-256 of the exact + # bytes, and every later read re-verifies that digest. if not bytes(prefix).startswith(b"%PDF-"): _raise(IpoDocumentDownloadErrorCode.INVALID_PDF) handle.flush() @@ -391,6 +382,12 @@ def _stream_pdf_to_cache( content_sha256 = digest.hexdigest() relative = PurePosixPath("ipo", "documents", f"{content_sha256}.pdf") final_path = _contained_cache_path(data_dir, relative.as_posix()) + # The atomic publish keeps no explicit file permissions (IPO-006 review + # note): the cached PDF inherits the temp file's mode, and + # confidentiality relies on the runtime data_dir itself being + # protected. That is the deployment model documented in + # docs/operations.md — these are public SEBI filings, so the threat is + # tampering (covered by the digest re-verification), not disclosure. os.replace(temporary_path, final_path) temporary_path = None return IpoDocumentDownloadResult( diff --git a/backend/ipo/sources/sebi.py b/backend/ipo/sources/sebi.py index 0d65f56..aabffc0 100644 --- a/backend/ipo/sources/sebi.py +++ b/backend/ipo/sources/sebi.py @@ -16,7 +16,6 @@ from collections.abc import Callable from dataclasses import dataclass from typing import Any -from urllib.parse import urljoin, urlsplit, urlunsplit import requests from bs4 import BeautifulSoup @@ -29,6 +28,7 @@ SebiFiling, SebiFilingCategory, ) +from backend.ipo.url_canonical import canonical_sebi_url AJAX_URL = "https://www.sebi.gov.in/sebiweb/ajax/home/getnewslistinfo.jsp" MAX_RESPONSE_BYTES = 2 * 1024 * 1024 @@ -80,19 +80,22 @@ def _canonical_sebi_url(value: str, *, base_url: str | None = None) -> str: The exact host, credential, scheme, and port checks are repeated for listing links and redirects. Removing fragments also ensures the record fingerprint identifies a server resource rather than browser-only navigation state. + + The implementation is shared with the prospectus downloader + (``backend/ipo/url_canonical.py``, IPO-006); this wrapper binds the listing + module's AJAX base URL, host allowlist, and error type. One deliberate + tightening rides along: a malformed port (``https://host:abc/``) now raises + ``SebiSourceError`` instead of leaking a bare ``ValueError`` — rejected + either way, but now within this module's error taxonomy. """ - candidate = urljoin(base_url or AJAX_URL, value.strip()) - parsed = urlsplit(candidate) - host = (parsed.hostname or "").casefold() - if ( - parsed.scheme.casefold() != "https" - or host not in ALLOWED_HOSTS - or parsed.username is not None - or parsed.password is not None - or parsed.port not in (None, 443) - ): - raise SebiSourceError("SEBI URL or redirect did not match the HTTPS host allowlist.") - return urlunsplit(("https", host, parsed.path or "/", parsed.query, "")) + return canonical_sebi_url( + value, + base_url=base_url or AJAX_URL, + allowed_hosts=ALLOWED_HOSTS, + error=lambda: SebiSourceError( + "SEBI URL or redirect did not match the HTTPS host allowlist." + ), + ) _SME_TOKEN = re.compile(r"(?:^|[\s(\[/{_-])SME(?:$|[\s)\]/}_-])", re.IGNORECASE) diff --git a/backend/ipo/url_canonical.py b/backend/ipo/url_canonical.py new file mode 100644 index 0000000..f400077 --- /dev/null +++ b/backend/ipo/url_canonical.py @@ -0,0 +1,101 @@ +"""One canonical SEBI URL gate for the two IPO fetch surfaces (IPO-006). + +Beginner note: +The listing scraper (``backend/ipo/sources/sebi.py``) and the prospectus +downloader (``backend/ipo/documents/downloader.py``) each canonicalized and +allowlisted SEBI URLs with a private copy of the same logic. The copies had +already drifted: the downloader's gained a malformed-port guard, an optional +DNS-answer check (against a poisoned hosts file or DNS response pointing +SEBI's name at private infrastructure), and a PDF-path restriction that the +scraper's copy lacked. This module is the single implementation; each caller +keeps a thin private wrapper that binds its own base URL, error type, and +hardening options, so both modules' public behavior and test suites stay +unchanged. +""" + +from __future__ import annotations + +import ipaddress +import socket +from collections.abc import Callable +from typing import Any, Never +from urllib.parse import urljoin, urlsplit, urlunsplit + + +def _reject(error: Callable[[], Exception]) -> Never: + """Raise the caller's error from a helper frame. + + Raising via a call (rather than a ``raise`` statement lexically inside an + ``except`` clause) preserves the implicit exception context exactly the + way both pre-IPO-006 copies did with their own raiser helpers. + """ + raise error() + + +def canonical_sebi_url( + value: str, + *, + base_url: str, + allowed_hosts: frozenset[str], + error: Callable[[], Exception], + resolver: Callable[..., Any] | None = None, + require_pdf_path: bool = False, +) -> str: + """Canonicalize one official SEBI HTTPS URL or raise the caller's error. + + Every rejection path raises ``error()`` so each caller keeps its own error + taxonomy (``SebiSourceError`` with a fixed message for the scraper, a + secret-safe ``unsafe_url`` code for the downloader) without this module + knowing either type. + + Checks, in order: + 1. Resolve ``value`` against ``base_url`` and split it. A malformed port + (e.g. ``https://host:abc/``) is rejected rather than leaking a bare + ``ValueError`` to the caller. + 2. Require exactly ``https``, a host in ``allowed_hosts``, no embedded + credentials, and port 443 (or none). + 3. With ``require_pdf_path=True``, additionally require SEBI's attachment + prefix (``/sebi_data/attachdocs/``) — the downloader's PDF fetches only. + 4. With a ``resolver`` (the downloader passes ``socket.getaddrinfo`` or a + test double; the scraper passes none), resolve the allowlisted host and + reject any non-public answer. Host allowlisting alone blocks ordinary + SSRF; this extra step also catches DNS answers that point the trusted + name at loopback/private ranges. + + The result drops the fragment and normalizes an empty path to ``/`` so a + record fingerprint identifies a server resource rather than browser-only + navigation state. + """ + candidate = urljoin(base_url, str(value).strip()) + parsed = urlsplit(candidate) + host = (parsed.hostname or "").casefold() + try: + port = parsed.port + except ValueError: + _reject(error) + if ( + parsed.scheme.casefold() != "https" + or host not in allowed_hosts + or parsed.username is not None + or parsed.password is not None + or port not in (None, 443) + ): + raise error() + if require_pdf_path and not parsed.path.startswith("/sebi_data/attachdocs/"): + raise error() + + if resolver is not None: + try: + answers = resolver(host, 443, type=socket.SOCK_STREAM) + addresses = {str(answer[4][0]) for answer in answers} + # An empty answer set or ANY non-global address fails closed; the + # ip_address() parse itself is inside the try so a malformed + # resolver answer is rejected, not raised. + unsafe = not addresses or any( + not ipaddress.ip_address(address).is_global for address in addresses + ) + except (OSError, TypeError, ValueError, IndexError): + _reject(error) + if unsafe: + raise error() + return urlunsplit(("https", host, parsed.path or "/", parsed.query, "")) diff --git a/tests/test_ipo_url_canonical.py b/tests/test_ipo_url_canonical.py new file mode 100644 index 0000000..5b22d25 --- /dev/null +++ b/tests/test_ipo_url_canonical.py @@ -0,0 +1,174 @@ +"""Direct tests for the shared SEBI URL canonicalizer (IPO-006). + +The listing scraper and the prospectus downloader exercise this function +end-to-end through their own suites (which pass unmodified); these tests lock +the shared implementation's knobs directly — the error factory, the optional +DNS answer check, and the PDF-path restriction — so a future edit cannot +weaken one caller's hardening without a failure here. +""" + +from __future__ import annotations + +import pytest + +from backend.ipo.url_canonical import canonical_sebi_url + +_ALLOWED = frozenset({"sebi.gov.in", "www.sebi.gov.in"}) +_BASE = "https://www.sebi.gov.in/sebiweb/ajax/home/getnewslistinfo.jsp" + + +class _Rejected(Exception): + """Caller-supplied error type; the canonicalizer must raise exactly this.""" + + +def _canonical(value: str, **overrides): + """Call the canonicalizer with this suite's defaults, overriding per test. + + Bundling ``base_url``/``allowed_hosts``/``error`` here keeps each test + focused on the one knob it varies, the way the two production wrappers + bind their own fixed configuration. + """ + kwargs = { + "base_url": _BASE, + "allowed_hosts": _ALLOWED, + "error": _Rejected, + } + kwargs.update(overrides) + return canonical_sebi_url(value, **kwargs) + + +def _resolver_answers(*addresses: str): + """Build a getaddrinfo-shaped resolver returning the given IP answers.""" + + def _resolver(_host, _port, **_kwargs): + """Return one socket-address tuple per programmed answer.""" + return [(None, None, None, None, (address, 443)) for address in addresses] + + return _resolver + + +# --------------------------------------------------------------------------- +# Canonicalization +# --------------------------------------------------------------------------- + + +def test_absolute_url_keeps_query_drops_fragment(): + """The fragment is browser-only state and must not reach fingerprints.""" + url = "https://www.sebi.gov.in/filings/jun-2026/demo.html?x=1#section-3" + assert _canonical(url) == "https://www.sebi.gov.in/filings/jun-2026/demo.html?x=1" + + +def test_relative_url_resolves_against_base(): + """Listing pages emit relative hrefs; they resolve against the caller's base.""" + assert _canonical("/filings/demo.html") == "https://www.sebi.gov.in/filings/demo.html" + + +def test_empty_path_normalizes_to_slash(): + """A bare host canonicalizes with an explicit root path.""" + assert _canonical("https://sebi.gov.in") == "https://sebi.gov.in/" + + +def test_host_casefolds_and_explicit_443_is_dropped(): + """Mixed-case hosts and an explicit default port collapse to one form.""" + assert ( + _canonical("https://WWW.SEBI.GOV.IN:443/doc.pdf") + == "https://www.sebi.gov.in/doc.pdf" + ) + + +# --------------------------------------------------------------------------- +# Rejections — every path raises the CALLER's error type +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "url", + [ + "http://www.sebi.gov.in/doc.pdf", # not https + "https://evil.example.com/doc.pdf", # host not allowlisted + "https://user@www.sebi.gov.in/doc.pdf", # embedded credentials + "https://user:pass@www.sebi.gov.in/doc.pdf", # embedded credentials + "https://www.sebi.gov.in:8443/doc.pdf", # non-443 port + "https://www.sebi.gov.in:abc/doc.pdf", # malformed port + ], +) +def test_unsafe_urls_raise_the_callers_error(url): + """Every rejection surfaces as the injected error type, never a bare one. + + The malformed-port case matters most: ``urlsplit(...).port`` raises a raw + ``ValueError``, and before IPO-006 the scraper's copy leaked it uncaught. + """ + with pytest.raises(_Rejected): + _canonical(url) + + +def test_require_pdf_path_restricts_to_attachdocs(): + """The downloader-only knob confines PDF fetches to SEBI's attachment tree.""" + good = "https://www.sebi.gov.in/sebi_data/attachdocs/jun-2026/demo.pdf" + assert _canonical(good, require_pdf_path=True).endswith("/demo.pdf") + + with pytest.raises(_Rejected): + _canonical("https://www.sebi.gov.in/other/demo.pdf", require_pdf_path=True) + # The same URL passes when the caller (the listing scraper) does not + # request the restriction. + assert _canonical("https://www.sebi.gov.in/other/demo.pdf") + + +# --------------------------------------------------------------------------- +# Optional DNS answer check (the downloader's anti-rebinding layer) +# --------------------------------------------------------------------------- + + +def test_resolver_accepting_public_answers_passes(): + """A host resolving only to public addresses is allowed through.""" + url = _canonical( + "https://www.sebi.gov.in/doc.pdf", resolver=_resolver_answers("1.2.3.4") + ) + assert url == "https://www.sebi.gov.in/doc.pdf" + + +@pytest.mark.parametrize("address", ["127.0.0.1", "10.0.0.5", "192.168.1.7", "::1"]) +def test_resolver_returning_private_answer_fails_closed(address): + """Loopback/private answers mean a poisoned resolution — reject the fetch.""" + with pytest.raises(_Rejected): + _canonical("https://www.sebi.gov.in/doc.pdf", resolver=_resolver_answers(address)) + + +def test_one_private_answer_among_public_ones_still_fails_closed(): + """A single private answer taints the whole set: ANY, not ALL, rejects.""" + with pytest.raises(_Rejected): + _canonical( + "https://www.sebi.gov.in/doc.pdf", + resolver=_resolver_answers("1.2.3.4", "127.0.0.1"), + ) + + +def test_empty_or_failing_resolver_fails_closed(): + """No answers and resolver errors both reject rather than proceeding blind.""" + with pytest.raises(_Rejected): + _canonical("https://www.sebi.gov.in/doc.pdf", resolver=_resolver_answers()) + + def _broken_resolver(_host, _port, **_kwargs): + """Model a DNS outage: getaddrinfo raising ``OSError``.""" + raise OSError("resolution failed") + + with pytest.raises(_Rejected): + _canonical("https://www.sebi.gov.in/doc.pdf", resolver=_broken_resolver) + + +def test_malformed_resolver_answer_fails_closed(): + """An answer that does not parse as an IP address rejects, not raises.""" + with pytest.raises(_Rejected): + _canonical( + "https://www.sebi.gov.in/doc.pdf", resolver=_resolver_answers("not-an-ip") + ) + + +def test_no_resolver_skips_the_dns_check_entirely(): + """resolver=None (the scraper's configuration) must skip DNS, not null-run it. + + An empty answer set fails closed WITH a resolver (asserted above), so this + passing proves the DNS layer is skipped rather than run with no answers. + """ + url = _canonical("https://www.sebi.gov.in/doc.pdf", resolver=None) + assert url == "https://www.sebi.gov.in/doc.pdf" From 6782cceb9d8855cf1562ec69f35f740e48eb89fd Mon Sep 17 00:00:00 2001 From: DoRmAmMu1997 Date: Sat, 11 Jul 2026 18:45:35 +0530 Subject: [PATCH 2/2] fix: harden SEBI URL canonicalization Co-authored-by: Hemant Co-authored-by: Codex --- backend/ipo/documents/downloader.py | 88 ++++++++-- backend/ipo/url_canonical.py | 66 ++++++- tests/test_ipo_document_downloader.py | 239 ++++++++++++++++++++++++++ tests/test_ipo_url_canonical.py | 35 ++++ 4 files changed, 409 insertions(+), 19 deletions(-) diff --git a/backend/ipo/documents/downloader.py b/backend/ipo/documents/downloader.py index 433da97..e8e1aee 100644 --- a/backend/ipo/documents/downloader.py +++ b/backend/ipo/documents/downloader.py @@ -25,7 +25,7 @@ from dataclasses import dataclass from pathlib import Path, PurePosixPath from typing import Any, Never -from urllib.parse import parse_qs, urljoin, urlsplit +from urllib.parse import parse_qs, urlsplit import requests from bs4 import BeautifulSoup @@ -139,9 +139,19 @@ def _request_with_redirects( url: str, *, resolver: Callable[..., Any], + require_pdf_path: bool = False, ) -> Any: - """GET one URL while validating and closing every manual redirect hop.""" - current_url = _canonical_sebi_url(url, resolver=resolver) + """GET one URL while validating and closing every manual redirect hop. + + ``require_pdf_path`` is deliberately carried through the whole redirect + chain. Validating only the iframe's first PDF URL would let a later 302 + leave SEBI's attachment directory after the stricter check had passed. + """ + current_url = _canonical_sebi_url( + url, + resolver=resolver, + require_pdf_path=require_pdf_path, + ) for redirect_count in range(MAX_REDIRECTS + 1): response = session.get( current_url, @@ -151,13 +161,22 @@ def _request_with_redirects( headers={"User-Agent": "Streamlit-Scanner-App/IPO-003"}, ) if response.status_code not in {301, 302, 303, 307, 308}: + # ``requests.Response.url`` normally carries this value, but test + # doubles and other requests-compatible sessions are not required + # to expose it. Preserve the exact URL that passed our redirect + # policy so the caller can apply the stricter PDF-path rule after + # it learns the response media type. + response._scanner_canonical_url = current_url return response try: location = response.headers.get("Location") if not location or redirect_count >= MAX_REDIRECTS: _raise(IpoDocumentDownloadErrorCode.UNSAFE_URL) current_url = _canonical_sebi_url( - str(location), base_url=current_url, resolver=resolver + str(location), + base_url=current_url, + resolver=resolver, + require_pdf_path=require_pdf_path, ) finally: response.close() @@ -170,12 +189,18 @@ def _fetch( *, resolver: Callable[..., Any], sleeper: Callable[[float], None], + require_pdf_path: bool = False, ) -> Any: """Return an open successful response after bounded transient retries.""" for attempt in range(len(RETRY_DELAYS_SECONDS) + 1): response = None try: - response = _request_with_redirects(session, url, resolver=resolver) + response = _request_with_redirects( + session, + url, + resolver=resolver, + require_pdf_path=require_pdf_path, + ) if response.status_code == 429 or 500 <= response.status_code <= 599: if attempt == len(RETRY_DELAYS_SECONDS): response.close() @@ -227,7 +252,19 @@ def _extract_pdf_url( source = iframe.get("src") if not source: continue - wrapper = urlsplit(urljoin(detail_url, str(source))) + try: + # The wrapper itself is untrusted page syntax even though we never + # fetch it. Canonicalizing it first catches a malformed host/port + # instead of extracting a valid-looking ``file`` parameter from an + # invalid URL. + wrapper_url = _canonical_sebi_url( + str(source), + base_url=detail_url, + resolver=resolver, + ) + wrapper = urlsplit(wrapper_url) + except (IpoDocumentDownloadError, TypeError, UnicodeError, ValueError): + _raise(IpoDocumentDownloadErrorCode.INVALID_DETAIL_PAGE) values = parse_qs(wrapper.query, keep_blank_values=True).get("file", []) if len(values) != 1 or not values[0].strip(): continue @@ -371,9 +408,10 @@ def _stream_pdf_to_cache( # Header-only PDF validation is deliberate (IPO-006 review note): # the magic-byte check rejects HTML error pages served with a PDF # content type, while deep structural validation is delegated to - # the parse stage. A truncated or corrupted body is still caught — - # the content-addressed cache stores the SHA-256 of the exact - # bytes, and every later read re-verifies that digest. + # the parse stage. The SHA-256 digest cannot prove that the source + # server sent a complete PDF; it detects only later alteration of + # the exact bytes stored here. Structural truncation is therefore + # the parser's responsibility. if not bytes(prefix).startswith(b"%PDF-"): _raise(IpoDocumentDownloadErrorCode.INVALID_PDF) handle.flush() @@ -449,15 +487,38 @@ def download_document_file( ) media_type = _content_type(response) if media_type == "text/html": + # Redirects can move a detail page to another directory. Resolve + # relative iframe and ``file`` values against the final URL that + # produced this HTML, not the stale URL requested before redirects. + final_detail_url = str( + getattr(response, "_scanner_canonical_url", detail_url) + ) try: pdf_url = _extract_pdf_url( - _read_html(response), detail_url=detail_url, resolver=resolver + _read_html(response), + detail_url=final_detail_url, + resolver=resolver, ) finally: response.close() response = None response = _fetch( - active_session, pdf_url, resolver=resolver, sleeper=sleeper + active_session, + pdf_url, + resolver=resolver, + sleeper=sleeper, + require_pdf_path=True, + ) + else: + # A listing URL may already return the prospectus PDF, but the PDF + # still has to live in SEBI's attachment tree. Media type alone must + # not promote an unrelated same-host resource into trusted filing + # evidence. The private attribute is set by our redirect loop to the + # final URL that was actually requested. + _canonical_sebi_url( + str(getattr(response, "_scanner_canonical_url", detail_url)), + resolver=resolver, + require_pdf_path=True, ) return _stream_pdf_to_cache( response, @@ -465,6 +526,11 @@ def download_document_file( data_dir=data_dir, downloaded_at=now().astimezone(dt.UTC), ) + except requests.RequestException: + # requests can raise lazily while ``iter_content`` reads an otherwise + # successful response. Convert those late failures to the same stable, + # secret-free taxonomy as connection failures raised by ``session.get``. + _raise(IpoDocumentDownloadErrorCode.NETWORK_ERROR) finally: if response is not None: response.close() diff --git a/backend/ipo/url_canonical.py b/backend/ipo/url_canonical.py index f400077..4f8f26c 100644 --- a/backend/ipo/url_canonical.py +++ b/backend/ipo/url_canonical.py @@ -16,10 +16,15 @@ from __future__ import annotations import ipaddress +import posixpath +import re import socket from collections.abc import Callable from typing import Any, Never -from urllib.parse import urljoin, urlsplit, urlunsplit +from urllib.parse import unquote, urljoin, urlsplit, urlunsplit + +_ENCODED_PATH_SEPARATOR = re.compile(r"%(?:2f|5c)", re.IGNORECASE) +_MAX_PERCENT_DECODING_PASSES = 4 def _reject(error: Callable[[], Exception]) -> Never: @@ -32,6 +37,45 @@ def _reject(error: Callable[[], Exception]) -> Never: raise error() +def _validate_pdf_path(path: str, error: Callable[[], Exception]) -> None: + """Require one unambiguous path inside SEBI's attachment directory. + + Beginner note: + ``urlsplit`` intentionally leaves percent escapes untouched. A downstream + HTTP client, proxy, or web server may decode them later, so a raw prefix + check is not enough: ``%2e%2e`` becomes ``..`` and ``%2f`` becomes ``/``. + We decode repeatedly to expose single- and double-encoded traversal, reject + encoded separators at every layer, and then compare normalized segments. + """ + decoded = path + for _pass in range(_MAX_PERCENT_DECODING_PASSES): + # Backslashes are separators on some servers. Percent-encoded slashes + # and backslashes are rejected rather than decoded because otherwise + # different network layers could disagree about the segment boundary. + if "\\" in decoded or _ENCODED_PATH_SEPARATOR.search(decoded): + raise error() + try: + next_value = unquote(decoded, errors="strict") + except (UnicodeDecodeError, ValueError): + _reject(error) + if next_value == decoded: + break + decoded = next_value + else: + # Excessively nested encodings are ambiguous and have no legitimate + # use in an official attachment path, so fail closed. + raise error() + + segments = decoded.split("/") + if ( + len(segments) < 4 + or segments[:3] != ["", "sebi_data", "attachdocs"] + or any(segment in {".", ".."} for segment in segments) + or posixpath.normpath(decoded) != decoded + ): + raise error() + + def canonical_sebi_url( value: str, *, @@ -66,12 +110,15 @@ def canonical_sebi_url( record fingerprint identifies a server resource rather than browser-only navigation state. """ - candidate = urljoin(base_url, str(value).strip()) - parsed = urlsplit(candidate) - host = (parsed.hostname or "").casefold() try: + # Keep every stdlib parsing accessor inside this boundary. ``urljoin`` + # and ``urlsplit`` can reject malformed bracketed hosts or Unicode + # netlocs, while ``hostname`` and ``port`` perform additional checks. + candidate = urljoin(base_url, str(value).strip()) + parsed = urlsplit(candidate) + host = (parsed.hostname or "").casefold() port = parsed.port - except ValueError: + except (TypeError, UnicodeError, ValueError): _reject(error) if ( parsed.scheme.casefold() != "https" @@ -81,8 +128,8 @@ def canonical_sebi_url( or port not in (None, 443) ): raise error() - if require_pdf_path and not parsed.path.startswith("/sebi_data/attachdocs/"): - raise error() + if require_pdf_path: + _validate_pdf_path(parsed.path, error) if resolver is not None: try: @@ -98,4 +145,7 @@ def canonical_sebi_url( _reject(error) if unsafe: raise error() - return urlunsplit(("https", host, parsed.path or "/", parsed.query, "")) + try: + return urlunsplit(("https", host, parsed.path or "/", parsed.query, "")) + except (TypeError, UnicodeError, ValueError): + _reject(error) diff --git a/tests/test_ipo_document_downloader.py b/tests/test_ipo_document_downloader.py index 3e8b31c..5bd9db5 100644 --- a/tests/test_ipo_document_downloader.py +++ b/tests/test_ipo_document_downloader.py @@ -59,6 +59,16 @@ def close(self) -> None: self.closed = True +class FailingStreamResponse(FakeResponse): + """Response double whose body fails after headers have been accepted.""" + + def iter_content(self, chunk_size: int) -> Iterator[bytes]: + """Raise lazily, matching failures surfaced by requests while streaming.""" + self.iterated = True + raise requests.ConnectionError("upstream token=supersecret123456") + yield b"" # pragma: no cover - keeps this method an iterator. + + class FakeSession: """FIFO request double used to prove redirects and retries deterministically.""" @@ -156,6 +166,48 @@ def test_direct_official_pdf_response_skips_html_resolution(tmp_path: Path) -> N assert len(session.calls) == 1 +def test_direct_pdf_response_must_use_the_official_attachment_path( + tmp_path: Path, +) -> None: + """A PDF media type must not turn an unrelated SEBI path into evidence.""" + session = FakeSession([FakeResponse(PDF_BYTES)]) + + with pytest.raises(IpoDocumentDownloadError) as caught: + download_document_file( + _document(document_url=DETAIL_URL), + data_dir=tmp_path, + session=session, + resolver=_public_resolver, + ) + + assert caught.value.code is IpoDocumentDownloadErrorCode.UNSAFE_URL + assert session.calls[0][0] == DETAIL_URL + + +@pytest.mark.parametrize("content_type", ["text/html", "application/pdf"]) +def test_stream_failure_uses_secret_safe_network_error_taxonomy( + tmp_path: Path, + content_type: str, +) -> None: + """Lazy body failures should look like every other safe network failure.""" + response = FailingStreamResponse(PDF_BYTES, content_type=content_type) + session = FakeSession([response]) + + with pytest.raises(IpoDocumentDownloadError) as caught: + download_document_file( + _document(document_url=PDF_URL), + data_dir=tmp_path, + session=session, + resolver=_public_resolver, + ) + + assert caught.value.code is IpoDocumentDownloadErrorCode.NETWORK_ERROR + assert "supersecret123456" not in str(caught.value) + assert response.closed + cache_dir = tmp_path / "ipo" / "documents" + assert not cache_dir.exists() or not list(cache_dir.iterdir()) + + def test_verified_cache_hit_performs_no_http_request(tmp_path: Path) -> None: """Rehash the stored file before declaring a zero-network cache hit.""" digest = hashlib.sha256(PDF_BYTES).hexdigest() @@ -188,6 +240,8 @@ def test_verified_cache_hit_performs_no_http_request(tmp_path: Path) -> None: "http://www.sebi.gov.in/filings/example.html", "https://user:password@www.sebi.gov.in/filings/example.html", "https://www.sebi.gov.in:444/filings/example.html", + "https://www.sebi.gov.in:invalid/filings/example.html", + "https://[www.sebi.gov.in/filings/example.html", "https://sebi.gov.in.evil.example/filings/example.html", ], ) @@ -340,6 +394,191 @@ def test_cross_host_redirect_is_rejected_and_response_is_closed(tmp_path: Path) assert redirect.closed +def test_pdf_redirect_cannot_escape_the_attachment_tree(tmp_path: Path) -> None: + """Keep the stricter PDF-path policy active on every redirect hop. + + The detail page may point at a valid attachment URL which then redirects. + Dropping ``require_pdf_path`` while following that redirect would let the + trusted host move the fetch to an unrelated path after validation. + """ + detail = FakeResponse( + f''.encode(), + content_type="text/html", + ) + redirect = FakeResponse( + b"", + status_code=302, + headers={"Location": "/sebi_data/private/secret.pdf"}, + ) + session = FakeSession([detail, redirect]) + + with pytest.raises(IpoDocumentDownloadError) as caught: + download_document_file( + _document(), + data_dir=tmp_path, + session=session, + resolver=_public_resolver, + ) + + assert caught.value.code is IpoDocumentDownloadErrorCode.UNSAFE_URL + assert [call[0] for call in session.calls] == [DETAIL_URL, PDF_URL] + assert detail.closed and redirect.closed + + +def test_pdf_redirect_within_attachment_tree_remains_allowed(tmp_path: Path) -> None: + """The redirect hardening must not block a normal in-tree PDF move.""" + redirected_pdf = "https://www.sebi.gov.in/sebi_data/attachdocs/final.pdf" + detail = FakeResponse( + f''.encode(), + content_type="text/html", + ) + redirect = FakeResponse( + b"", + status_code=302, + headers={"Location": redirected_pdf}, + ) + session = FakeSession([detail, redirect, FakeResponse(PDF_BYTES)]) + + result = download_document_file( + _document(), + data_dir=tmp_path, + session=session, + resolver=_public_resolver, + ) + + assert result.bytes_written == len(PDF_BYTES) + assert [call[0] for call in session.calls] == [DETAIL_URL, PDF_URL, redirected_pdf] + + +def test_redirected_detail_page_resolves_relative_iframe_from_final_url( + tmp_path: Path, +) -> None: + """Relative iframe paths belong to the page that actually returned HTML. + + Beginner note: a redirect can move a detail page into another directory. + Resolving its relative links against the original pre-redirect URL invents + a different resource and can reject a legitimate prospectus. + """ + redirected_detail = ( + "https://www.sebi.gov.in/sebi_data/attachdocs/2026/detail.html" + ) + relative_pdf = ( + "https://www.sebi.gov.in/sebi_data/attachdocs/2026/prospectus.pdf" + ) + redirect = FakeResponse( + b"", + status_code=302, + headers={"Location": redirected_detail}, + ) + detail = FakeResponse( + b'', + content_type="text/html", + ) + session = FakeSession([redirect, detail, FakeResponse(PDF_BYTES)]) + + result = download_document_file( + _document(), + data_dir=tmp_path, + session=session, + resolver=_public_resolver, + ) + + assert result.bytes_written == len(PDF_BYTES) + assert [call[0] for call in session.calls] == [ + DETAIL_URL, + redirected_detail, + relative_pdf, + ] + + +@pytest.mark.parametrize( + "location", + [ + "https://[www.sebi.gov.in/file.pdf", + "https://www.sebi.gov.in:invalid/file.pdf", + ], +) +def test_malformed_redirect_has_safe_url_error_and_closes_response( + tmp_path: Path, + location: str, +) -> None: + """Malformed redirect syntax stays inside the downloader error taxonomy.""" + redirect = FakeResponse(b"", status_code=302, headers={"Location": location}) + + with pytest.raises(IpoDocumentDownloadError) as caught: + download_document_file( + _document(), + data_dir=tmp_path, + session=FakeSession([redirect]), + resolver=_public_resolver, + ) + + assert caught.value.code is IpoDocumentDownloadErrorCode.UNSAFE_URL + assert redirect.closed + + +@pytest.mark.parametrize( + "iframe_source", + [ + "https://[www.sebi.gov.in/web/?file=/sebi_data/attachdocs/demo.pdf", + "https://www.sebi.gov.in:invalid/web/?file=/sebi_data/attachdocs/demo.pdf", + ], +) +def test_malformed_iframe_wrapper_has_invalid_detail_page_error( + tmp_path: Path, + iframe_source: str, +) -> None: + """Broken wrapper syntax is a bad detail page, not a raw parser exception.""" + detail = FakeResponse( + f''.encode(), + content_type="text/html", + ) + + with pytest.raises(IpoDocumentDownloadError) as caught: + download_document_file( + _document(), + data_dir=tmp_path, + session=FakeSession([detail]), + resolver=_public_resolver, + ) + + assert caught.value.code is IpoDocumentDownloadErrorCode.INVALID_DETAIL_PAGE + assert detail.closed + + +@pytest.mark.parametrize( + "iframe_file", + [ + "/sebi_data/attachdocs/../secret.pdf", + "/sebi_data/attachdocs/%2e%2e/secret.pdf", + "/sebi_data/attachdocs/%252e%252e/secret.pdf", + "/sebi_data/attachdocs/demo%252f..%252fsecret.pdf", + ], +) +def test_iframe_pdf_target_rejects_encoded_path_confusion_before_http( + tmp_path: Path, + iframe_file: str, +) -> None: + """A hostile iframe target must never become the download request.""" + detail = FakeResponse( + f''.encode(), + content_type="text/html", + ) + session = FakeSession([detail]) + + with pytest.raises(IpoDocumentDownloadError) as caught: + download_document_file( + _document(), + data_dir=tmp_path, + session=session, + resolver=_public_resolver, + ) + + assert caught.value.code is IpoDocumentDownloadErrorCode.UNSAFE_URL + assert [call[0] for call in session.calls] == [DETAIL_URL] + assert detail.closed + + def test_terminal_http_error_closes_response(tmp_path: Path) -> None: """Release the connection even when a non-retryable response cannot be used.""" response = FakeResponse(b"not found", status_code=404, content_type="text/html") diff --git a/tests/test_ipo_url_canonical.py b/tests/test_ipo_url_canonical.py index 5b22d25..13a4a93 100644 --- a/tests/test_ipo_url_canonical.py +++ b/tests/test_ipo_url_canonical.py @@ -90,6 +90,8 @@ def test_host_casefolds_and_explicit_443_is_dropped(): "https://user:pass@www.sebi.gov.in/doc.pdf", # embedded credentials "https://www.sebi.gov.in:8443/doc.pdf", # non-443 port "https://www.sebi.gov.in:abc/doc.pdf", # malformed port + "https://[www.sebi.gov.in/doc.pdf", # malformed bracketed host + "https://www.sebi.gov.in\uff0fevil/doc.pdf", # invalid NFKC netloc character ], ) def test_unsafe_urls_raise_the_callers_error(url): @@ -114,6 +116,39 @@ def test_require_pdf_path_restricts_to_attachdocs(): assert _canonical("https://www.sebi.gov.in/other/demo.pdf") +@pytest.mark.parametrize( + "unsafe_path", + [ + "/sebi_data/attachdocs/../secret.pdf", + "/sebi_data/attachdocs/%2e%2e/secret.pdf", + "/sebi_data/attachdocs/%252e%252e/secret.pdf", + "/sebi_data/attachdocs/demo%2f..%2fsecret.pdf", + "/sebi_data/attachdocs/demo%252f..%252fsecret.pdf", + "/sebi_data/attachdocs/demo%5c..%5csecret.pdf", + r"/sebi_data/attachdocs/demo\..\secret.pdf", + ], +) +def test_pdf_path_rejects_traversal_and_encoded_separators(unsafe_path): + """Decode path segments before accepting SEBI's attachment directory. + + Beginner note: + HTTP clients and reverse proxies can decode percent escapes at different + times. Checking only the raw string therefore lets ``%2e%2e`` (``..``) or + an encoded slash acquire a different meaning after this guard has run. + """ + with pytest.raises(_Rejected): + _canonical( + f"https://www.sebi.gov.in{unsafe_path}", + require_pdf_path=True, + ) + + +def test_malformed_base_url_raises_the_callers_error(): + """A bad join base is categorized instead of leaking ``ValueError``.""" + with pytest.raises(_Rejected): + _canonical("relative.pdf", base_url="https://[") + + # --------------------------------------------------------------------------- # Optional DNS answer check (the downloader's anti-rebinding layer) # ---------------------------------------------------------------------------