Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion docs/API.md
Original file line number Diff line number Diff line change
Expand Up @@ -611,7 +611,7 @@ Download previously exported results using a time-limited token.
X-API-Key: your-api-key-here
```

**Response:** File stream (application/octet-stream)
**Response:** File stream (application/octet-stream). The `Content-Disposition` header carries a filename matching the export format: `attachment; filename="export.json"` (also `.jsonl`, `.csv` or `.html`).

**Status Codes:**

Expand Down
29 changes: 27 additions & 2 deletions src/ciberwebscan/api/helpers/download_helper.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,17 +7,41 @@

from __future__ import annotations

from pathlib import Path
from typing import TypeVar
from urllib.parse import urlparse

from fastapi import Request

from ciberwebscan.export import EXPORT_FORMATS
from ciberwebscan.services.base import ServiceResult
from ciberwebscan.services.download_service import DownloadService

T = TypeVar("T")


def _detect_export_format(export_format: str | None, export_path: Path) -> str:
"""Resolve the export format for download metadata.

Preference order: the format reported by the service, then the export
file extension, then ``json``. Unknown values are discarded so they
can never surface in download metadata.

Args:
export_format: Format reported by the service result, if any.
export_path: Path of the exported file.

Returns:
One of ``EXPORT_FORMATS``.
"""
if export_format is not None and export_format in EXPORT_FORMATS:
return export_format
suffix = export_path.suffix.removeprefix(".").lower()
if suffix in EXPORT_FORMATS:
return suffix
return "json"


def enrich_response_with_token(
result: ServiceResult[T],
user_id: str,
Expand Down Expand Up @@ -46,11 +70,12 @@ def enrich_response_with_token(
if result.export_path is None:
return result.data, None, None

# Generate token from the exported file
# Generate token from the exported file with the real export format
# (previously hardcoded to "json", which mislabeled every download).
token_result = download_service.generate_download_token(
file_path=result.export_path,
user_id=user_id,
file_format="json", # Default format, could be detected from path extension
file_format=_detect_export_format(result.export_format, result.export_path),
)

# If token generation failed, log but don't error
Expand Down
12 changes: 7 additions & 5 deletions src/ciberwebscan/api/models/requests.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,8 @@

from pydantic import BaseModel, Field, HttpUrl, field_validator

from ciberwebscan.export import ExportFormat

# =============================================================================
# Scrape Requests
# =============================================================================
Expand Down Expand Up @@ -77,7 +79,7 @@ class ScrapeRequest(BaseModel):
default=None,
description="Optional output file path for exported results",
)
export_format: Literal["json", "jsonl", "csv", "html"] = Field(
export_format: ExportFormat = Field(
default="json",
description="Export format when export path is provided",
)
Expand Down Expand Up @@ -112,7 +114,7 @@ class ScrapeBatchRequest(BaseModel):
default=None,
description="Optional output file path for exported batch results",
)
export_format: Literal["json", "jsonl", "csv", "html"] = Field(
export_format: ExportFormat = Field(
default="jsonl",
description="Export format when export path is provided",
)
Expand Down Expand Up @@ -176,7 +178,7 @@ class AnalyzeRequest(BaseModel):
default=None,
description="Optional output file path for exported results",
)
export_format: Literal["json", "jsonl", "csv", "html"] = Field(
export_format: ExportFormat = Field(
default="json",
description="Export format when export path is provided",
)
Expand Down Expand Up @@ -265,7 +267,7 @@ class AttackRequest(BaseModel):
default=None,
description="Optional output file path for exported results",
)
export_format: Literal["json", "jsonl", "csv", "html"] = Field(
export_format: ExportFormat = Field(
default="json",
description="Export format when export path is provided",
)
Expand Down Expand Up @@ -319,7 +321,7 @@ class ExportRequest(BaseModel):
"""Request for exporting results."""

job_id: str = Field(..., description="Job ID from a previous operation")
format: Literal["jsonl", "json", "csv", "html"] = "jsonl"
format: ExportFormat = "jsonl"
include_raw_html: bool = False
include_screenshots: bool = False

Expand Down
2 changes: 1 addition & 1 deletion src/ciberwebscan/api/models/responses.py
Original file line number Diff line number Diff line change
Expand Up @@ -267,7 +267,7 @@ class DownloadInfo(BaseModel):
expires_at: datetime
attempts_remaining: int
file_format: str = Field(
default="json", description="Export format (json/jsonl/csv)"
default="json", description="Export format (json/jsonl/csv/html)"
)


Expand Down
31 changes: 30 additions & 1 deletion src/ciberwebscan/api/routes/download.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,13 +16,32 @@

from ciberwebscan.api.auth import AuthenticatedUser, get_current_user
from ciberwebscan.config.loader import get_config
from ciberwebscan.export import EXPORT_FORMATS
from ciberwebscan.services.download_service import DownloadService
from ciberwebscan.utils.logging import mask_identifier

logger = logging.getLogger(__name__)

router = APIRouter(prefix="/download", tags=["download"])


def _content_disposition(file_format: str | None) -> str:
"""Build the Content-Disposition header for an export format.

Only formats in ``EXPORT_FORMATS`` reach the header, so a
tampered registry value can never inject header content; anything
else falls back to ``json``.

Args:
file_format: Format stored in the token metadata, if any.

Returns:
The Content-Disposition value with a format-specific filename.
"""
extension = file_format if file_format in EXPORT_FORMATS else "json"
return f'attachment; filename="export.{extension}"'


# Service instance
_download_service = DownloadService()

Expand Down Expand Up @@ -228,9 +247,19 @@ async def download_file(
detail="Failed to retrieve file stream",
)

# Token metadata: the stored export format drives the download filename
info_result = download_service.get_info(token=token)
if not info_result.success or info_result.data is None:
reservation.refund_once()
logger.error(f"Failed to get download metadata: {info_result.error}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="Failed to retrieve file metadata",
)

# Prepare headers
headers = {
"Content-Disposition": "attachment; filename=export.json",
"Content-Disposition": _content_disposition(info_result.data.file_format),
"X-Content-Type-Options": "nosniff",
"Cache-Control": "no-cache, no-store, must-revalidate",
"Pragma": "no-cache",
Expand Down
5 changes: 5 additions & 0 deletions src/ciberwebscan/export/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -31,8 +31,10 @@

# Base classes and utilities
from ciberwebscan.export.base import (
EXPORT_FORMATS,
BaseExporter,
ExportError,
ExportFormat,
ExportValidationError,
ExportWriteError,
StreamingExporter,
Expand Down Expand Up @@ -102,6 +104,9 @@
"ExportError",
"ExportWriteError",
"ExportValidationError",
# Export formats
"ExportFormat",
"EXPORT_FORMATS",
# Factory functions
"export_to_file",
"get_exporter",
Expand Down
12 changes: 9 additions & 3 deletions src/ciberwebscan/export/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,13 +13,19 @@
from contextlib import contextmanager
from datetime import datetime
from pathlib import Path
from typing import TYPE_CHECKING, Any, TextIO
from typing import TYPE_CHECKING, Any, Literal, TextIO, get_args

if TYPE_CHECKING:
from ciberwebscan.export.models import AnalysisReport

logger = logging.getLogger(__name__)

# Canonical export formats. Single source of truth: the Literal drives
# validation across API models and download metadata, and EXPORT_FORMATS
# is derived from it for runtime membership checks.
ExportFormat = Literal["json", "jsonl", "csv", "html"]
EXPORT_FORMATS: frozenset[str] = frozenset(get_args(ExportFormat))


class ExportError(Exception):
"""Base exception for export errors."""
Expand Down Expand Up @@ -320,9 +326,9 @@ def get_exporter(
}

format_lower = format.lower()
if format_lower not in exporters:
if format_lower not in EXPORT_FORMATS:
raise ExportError(
f"Unknown export format: {format}. Use: {list(exporters.keys())}"
f"Unknown export format: {format}. Use: {sorted(EXPORT_FORMATS)}"
)

return exporters[format_lower](output_path=output_path, **kwargs)
26 changes: 25 additions & 1 deletion src/ciberwebscan/services/download_service.py
Original file line number Diff line number Diff line change
Expand Up @@ -141,7 +141,7 @@ def generate_download_token(
Args:
file_path: Path to file to download. Must be within allowed directories.
user_id: ID of user requesting download
file_format: Format of the exported file (json/jsonl/csv)
file_format: Format of the exported file (json/jsonl/csv/html)

Returns:
ServiceResult with DownloadTokenResponse containing token and expiration
Expand Down Expand Up @@ -329,6 +329,30 @@ def restore_attempt(self, token: str) -> ServiceResult[bool]:
self.logger.error(f"Error restoring attempt for token {token}: {e}")
return ServiceResult(success=False, error=str(e))

def get_info(self, token: str) -> ServiceResult[DownloadInfo]:
"""
Retrieve metadata (including file_format) for a download token.

Read-only: expiration, ownership and attempt checks are the
caller's responsibility (``validate_download_request``), which
runs before any use of this metadata.

Args:
token: Download token

Returns:
ServiceResult with the token's DownloadInfo, or a failure
result when the token no longer exists.
"""
try:
info = run_async(_registry.get_info(token))
if info is None:
return ServiceResult(success=False, error="Token not found")
return ServiceResult(success=True, data=info)
except Exception as e:
self.logger.error(f"Error getting token info: {e}")
return ServiceResult(success=False, error=str(e))

def get_file_stream(
self,
token: str,
Expand Down
36 changes: 36 additions & 0 deletions tests/integration/api/test_api_export_download.py
Original file line number Diff line number Diff line change
Expand Up @@ -265,6 +265,42 @@ def test_downloaded_html_has_valid_content(
assert "<h1>Report</h1>" in text


class TestDownloadContentDisposition:
"""The Content-Disposition filename must reflect the export format."""

@pytest.mark.parametrize(
("file_format", "file_name"),
[
("json", "export.json"),
("jsonl", "export.jsonl"),
("csv", "export.csv"),
("html", "export.html"),
],
)
def test_header_filename_matches_export_format(
self, in_process_client: tuple, tmp_path: Path, file_format: str, file_name: str
):
client, _ = in_process_client

test_file = tmp_path / f"data.{file_format}"
test_file.write_text("payload")

service = DownloadService()
result = service.generate_download_token(
file_path=test_file,
user_id="test_user",
file_format=file_format,
)
token = result.data.token

response = client.get(f"/api/download/{token}")
assert response.status_code == 200
assert (
response.headers["content-disposition"]
== f'attachment; filename="{file_name}"'
)


class TestTokenLifecycle:
"""End-to-end token lifecycle test."""

Expand Down
Empty file.
Loading
Loading