diff --git a/backend/ipo/documents/downloader.py b/backend/ipo/documents/downloader.py
index 5b60e7a..e8e1aee 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, 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:
@@ -154,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,
@@ -166,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()
@@ -185,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()
@@ -242,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
@@ -383,6 +405,13 @@ 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. 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()
@@ -391,6 +420,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(
@@ -452,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,
@@ -468,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/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..4f8f26c
--- /dev/null
+++ b/backend/ipo/url_canonical.py
@@ -0,0 +1,151 @@
+"""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 posixpath
+import re
+import socket
+from collections.abc import Callable
+from typing import Any, Never
+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:
+ """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 _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,
+ *,
+ 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.
+ """
+ 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 (TypeError, UnicodeError, 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:
+ _validate_pdf_path(parsed.path, 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()
+ 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
new file mode 100644
index 0000000..13a4a93
--- /dev/null
+++ b/tests/test_ipo_url_canonical.py
@@ -0,0 +1,209 @@
+"""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
+ "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):
+ """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")
+
+
+@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)
+# ---------------------------------------------------------------------------
+
+
+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"