Implement service authentication and error handling; update environment variables and documentation

This commit is contained in:
David Arranz 2026-09-24 10:12:42 +02:00
parent 5ab10cf6c8
commit 9913ec5241
16 changed files with 468 additions and 228 deletions

View File

@ -24,11 +24,12 @@ SECRET_PROVIDER=infisical # fake | google | infisical
# Infisical
INFISICAL_HOST=https://eu.infisical.com
INFISICAL_CLIENT_ID=35f83820-a9d3-4622-a0ab-ae6170f662fa
INFISICAL_CLIENT_SECRET=2c158d6f77fe4fc684bdb78d0c1ed21ed7a762885f508facc575aa05c42397c5
INFISICAL_PROJECT_ID=0bd3c2e0-39f5-4f92-8c6e-49bcc7eda896
INFISICAL_CLIENT_ID=replace-with-infisical-client-id
INFISICAL_CLIENT_SECRET=replace-with-infisical-client-secret
INFISICAL_PROJECT_ID=replace-with-infisical-project-id
INFISICAL_ENV_SLUG=dev
INFISICAL_TOKEN_AUTH=eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpZGVudGl0eUlkIjoiMmNjYmRkZmMtMGQyOC00ZmIwLTlhMGEtNWQyYTZiZWU3YzliIiwiaWRlbnRpdHlBY2Nlc3NUb2tlbklkIjoiZWMzNjg1OGUtNTkzYi00ZTE3LWI2ZmItN2NhODdlZDBhMmZhIiwiYXV0aFRva2VuVHlwZSI6ImlkZW50aXR5QWNjZXNzVG9rZW4iLCJpYXQiOjE3Njk2MTQ3NzQsImV4cCI6MTc3MjIwNjc3NH0.KZvzabbsgevhukPfc8LSlkOwos5eBaGvaqC2XPJEGHI
# Reserved for backwards compatibility; it is not used by the current SDK adapter.
INFISICAL_TOKEN_AUTH=replace-with-token-if-your-deployment-requires-it
# Google
GCP_PROJECT_ID=
@ -38,3 +39,8 @@ GCP_PROJECT_ID=
# --------------------
CERT_SECRET_NAME=certificate_secret_name
CERT_PASSWORD_SECRET_NAME=certificate_password_secret_name
# Technical client authentication for POST /documents/sign. Send it as:
# Authorization: Bearer <SERVICE_API_KEY>
SERVICE_API_KEY=replace-with-a-long-random-service-api-key
# Maximum accepted upload size. The service reads at most this value plus one byte.
PDF_MAX_SIZE_BYTES=10485760

View File

@ -0,0 +1,16 @@
# FactuGES Document Signing Service
## Authentication
`POST /documents/sign` requires a technical credential configured in
`SERVICE_API_KEY` and sent as `Authorization: Bearer <SERVICE_API_KEY>`.
`GET /health` and `GET /version` remain public. Set `PDF_MAX_SIZE_BYTES` to
limit in-memory PDF uploads (default: 10 MiB).
The multipart fields `certificate_secret_name`,
`certificate_password_secret_name`, and `company_slug` remain required for
backwards compatibility. Certificate selection continues to use
`CERT_SECRET_NAME` and `CERT_PASSWORD_SECRET_NAME`.
Any Infisical credentials that were previously committed must be revoked or
rotated in Infisical outside this repository.

View File

@ -29,6 +29,7 @@ dependencies = [
dev = [
"pytest>=8.0",
"pytest-cov",
"httpx",
"ruff",
"mypy",
]

View File

@ -0,0 +1,29 @@
"""Authentication dependencies for technical API consumers."""
import secrets
from typing import Annotated
from fastapi import Depends
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from signing_service.application.errors import (
AuthenticationRequiredError,
InvalidCredentialsError,
)
from signing_service.application.settings.container import get_settings
_bearer = HTTPBearer(auto_error=False)
async def require_service_authentication(
credentials: Annotated[
HTTPAuthorizationCredentials | None, Depends(_bearer)
] = None,
) -> None:
"""Require a configured Bearer token without exposing supplied credentials."""
if credentials is None or credentials.scheme.lower() != "bearer":
raise AuthenticationRequiredError()
expected = get_settings().service_api_key
if not secrets.compare_digest(credentials.credentials, expected):
raise InvalidCredentialsError()

View File

@ -1,35 +1,36 @@
from fastapi import APIRouter, UploadFile, File, Form, HTTPException
from typing import Annotated
from fastapi import APIRouter, Depends, File, Form, UploadFile
from fastapi.responses import Response
from signing_service.api.authentication import require_service_authentication
from signing_service.application.errors import PayloadTooLargeError
from signing_service.application.settings.container import get_settings
from signing_service.application.use_cases.sign_document import (
SignDocumentUseCase,
)
from signing_service.application.use_cases.sign_document_dto import (
SignDocumentCommand,
from signing_service.application.use_cases.sign_document import SignDocumentUseCase
from signing_service.application.use_cases.sign_document_dto import SignDocumentCommand
from signing_service.infrastructure.factories.pdf_signer_factory import (
create_pdf_signer,
)
from signing_service.infrastructure.factories.secret_manager_factory import (
create_secret_manager,
)
from signing_service.infrastructure.factories.pdf_signer_factory import (
create_pdf_signer,
)
router = APIRouter(prefix="/documents", tags=["documents"])
@router.post("/sign")
async def sign_document(
file: UploadFile = File(...),
certificate_secret_name: str = Form(...),
certificate_password_secret_name: str = Form(...),
company_slug: str = Form(...),
file: Annotated[UploadFile, File(...)],
certificate_secret_name: Annotated[str, Form(...)],
certificate_password_secret_name: Annotated[str, Form(...)],
company_slug: Annotated[str, Form(...)],
_: Annotated[None, Depends(require_service_authentication)],
):
try:
"""Sign with the globally configured legacy certificate selection."""
settings = get_settings()
pdf_bytes = await file.read()
pdf_bytes = await file.read(settings.pdf_max_size_bytes + 1)
if len(pdf_bytes) > settings.pdf_max_size_bytes:
raise PayloadTooLargeError()
use_case = SignDocumentUseCase(
secret_manager=create_secret_manager(),
@ -37,23 +38,16 @@ async def sign_document(
cert_secret_name=settings.pdf_cert_secret_name,
cert_password_secret_name=settings.pdf_cert_password_secret_name,
)
command = SignDocumentCommand(
pdf_bytes=pdf_bytes,
certificate_secret_name=certificate_secret_name,
certificate_password_secret_name=certificate_password_secret_name,
company_slug=company_slug,
)
result = await use_case.execute(command)
return Response(
content=result.signed_pdf_bytes,
media_type="application/pdf",
headers={
"Content-Disposition": f"attachment; filename=signed_{file.filename}"
},
headers={"Content-Disposition": f"attachment; filename=signed_{file.filename}"},
)
except Exception as exc:
raise HTTPException(status_code=400, detail=str(exc))

View File

@ -0,0 +1,43 @@
"""Stable, safe errors exposed by the signing service."""
class ServiceError(Exception):
code = "SIGNING_FAILED"
status_code = 500
message = "Document signing failed"
class AuthenticationRequiredError(ServiceError):
code = "AUTHENTICATION_REQUIRED"
status_code = 401
message = "Authentication is required"
class InvalidCredentialsError(ServiceError):
code = "INVALID_CREDENTIALS"
status_code = 401
message = "Invalid credentials"
class InvalidRequestError(ServiceError):
code = "INVALID_REQUEST"
status_code = 400
message = "Invalid request"
class InvalidPdfError(ServiceError):
code = "INVALID_PDF"
status_code = 400
message = "Invalid PDF document"
class PayloadTooLargeError(ServiceError):
code = "INVALID_REQUEST"
status_code = 413
message = "PDF exceeds the configured maximum size"
class SecretProviderError(ServiceError):
code = "SECRET_PROVIDER_ERROR"
status_code = 503
message = "Secret provider is unavailable"

View File

@ -10,6 +10,8 @@ class AppSettings:
log_level: str
state_path: str
local_tz: str
service_api_key: str
pdf_max_size_bytes: int
# --------------------
# Secret manager

View File

@ -56,6 +56,16 @@ def load_app_settings() -> AppSettings:
default="UTC",
)
service_api_key = get_env("SERVICE_API_KEY", required=True)
pdf_max_size_raw = get_env("PDF_MAX_SIZE_BYTES", default="10485760")
try:
pdf_max_size_bytes = int(pdf_max_size_raw)
except (TypeError, ValueError) as exc:
raise ValueError("PDF_MAX_SIZE_BYTES must be a positive integer") from exc
if pdf_max_size_bytes <= 0:
raise ValueError("PDF_MAX_SIZE_BYTES must be a positive integer")
# --------------------
# Secret manager
# --------------------
@ -96,10 +106,8 @@ def load_app_settings() -> AppSettings:
required=secret_provider == "infisical",
)
infisical_token_auth = get_env(
"INFISICAL_TOKEN_AUTH",
required=secret_provider == "infisical",
)
# Kept for configuration compatibility. The current universal-auth adapter does not use it.
infisical_token_auth = get_env("INFISICAL_TOKEN_AUTH")
gcp_project_id = get_env(
"GCP_PROJECT_ID",
@ -127,6 +135,8 @@ def load_app_settings() -> AppSettings:
log_level=log_level,
state_path=state_path,
local_tz=local_tz,
service_api_key=service_api_key,
pdf_max_size_bytes=pdf_max_size_bytes,
secret_provider=secret_provider,
infisical_host=infisical_host,
infisical_client_id=infisical_client_id,

View File

@ -1,9 +1,11 @@
from signing_service.application.errors import SecretProviderError, ServiceError
from signing_service.application.use_cases.sign_document_dto import (
SignDocumentCommand,
SignDocumentResult,
)
from signing_service.domain.ports.secret_manager import SecretManagerPort
from signing_service.domain.ports.pdf_signer import PDFSignerPort
from signing_service.domain.ports.secret_manager import SecretManagerPort
from signing_service.infrastructure.pdf.pdf_validator_error import PDFValidationError
class SignDocumentUseCase:
@ -20,18 +22,25 @@ class SignDocumentUseCase:
self._cert_password_secret_name = cert_password_secret_name
async def execute(self, command: SignDocumentCommand) -> SignDocumentResult:
certificate = self._secret_manager.get_secret(
self._cert_secret_name
)
password = self._secret_manager.get_secret(
self._cert_password_secret_name
)
try:
certificate = self._secret_manager.get_secret(self._cert_secret_name)
password = self._secret_manager.get_secret(self._cert_password_secret_name)
except Exception as exc:
raise SecretProviderError() from exc
try:
signed_pdf = await self._pdf_signer.sign(
pdf_bytes=command.pdf_bytes,
certificate=certificate,
password=password,
)
except PDFValidationError as exc:
from signing_service.application.errors import InvalidPdfError
raise InvalidPdfError() from exc
except ServiceError:
raise
except Exception as exc:
raise ServiceError() from exc
return SignDocumentResult(signed_pdf_bytes=signed_pdf)

View File

@ -22,9 +22,6 @@ def create_secret_manager() -> SecretManagerPort:
if not settings.infisical_project_id:
raise ValueError("INFISICAL_PROJECT_ID is required for Infisical Secret Manager")
if not settings.infisical_token_auth:
raise ValueError("INFISICAL_TOKEN_AUTH is required for Infisical Secret Manager")
if not settings.infisical_env_slug:
raise ValueError("INFISICAL_ENV_SLUG is required for Infisical Secret Manager")

View File

@ -2,15 +2,7 @@ from signing_service.domain.ports.pdf_signer import PDFSignerPort
class FakePDFSigner(PDFSignerPort):
"""
Fake implementation of PDFSignerPort.
It does NOT sign a real PDF.
It only simulates the behavior for testing and development.
"""
"""Test/development signer that never includes secret material in output."""
async def sign(self, pdf_bytes: bytes, certificate: str, password: str) -> bytes:
# Simulación simple:
# añadimos un marcador para indicar "firmado"
signature_marker = f"\n\n-- Signed with certificate: {certificate} and password: {password}".encode()
return pdf_bytes + signature_marker
return pdf_bytes + b"\n\n-- Signed by fake signer"

View File

@ -1,133 +1,121 @@
from contextlib import asynccontextmanager
from datetime import UTC, datetime
from pathlib import Path
from fastapi import FastAPI, HTTPException, Request
from dotenv import load_dotenv
from fastapi import FastAPI, Request
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from starlette.middleware.base import BaseHTTPMiddleware
from dotenv import load_dotenv
from pathlib import Path
from datetime import datetime, UTC
from dateutil import tz
from signing_service.api.routes.sign_document import router as sign_router
from signing_service.application.errors import InvalidRequestError, ServiceError
from signing_service.application.settings.container import get_settings
from signing_service.application.settings.setup_logger import create_logger
from signing_service.api.routes.sign_document import router as sign_router
from signing_service.application.settings.version import get_package_version
load_dotenv()
# 👇 FAIL FAST: load settings at startup
settings = get_settings()
version = get_package_version()
def build_logger():
"""Crea el logger de aplicación y asegura el directorio de logs."""
log_dir = Path(settings.state_path)
log_dir.mkdir(parents=True, exist_ok=True)
app_logger = create_logger(
name="factuges-document-signing-service",
log_path=log_dir / "factuges-document-signing-service.log",
)
app_logger.info("")
app_logger.info(
"============================================================")
app_logger.info("FactuGES Document Signing Service - START")
app_logger.info("FactuGES Document Signing Service started")
app_logger.info("Version: %s", version)
app_logger.info("UTC Now: %s", datetime.now(UTC).isoformat())
app_logger.info("Environment: %s", settings.app_env)
app_logger.info("")
app_logger.info("Log Level: %s", settings.log_level)
app_logger.info(
"Log: %s",
log_dir / "factuges-document-signing-service.log",
)
app_logger.info("")
app_logger.info("Secret Provider: %s", settings.secret_provider)
app_logger.info("Infisical project ID: %s", settings.infisical_project_id)
app_logger.info("Infisical environment: %s", settings.infisical_env_slug)
app_logger.info("")
app_logger.info("Secret provider configured: %s", settings.secret_provider)
return app_logger
logger = build_logger()
async def http_exception_handler(request: Request, exc: HTTPException):
"""Registra y devuelve errores HTTP controlados."""
logger.error(
"HTTPException %s %s | status=%s | detail=%s",
request.method,
request.url.path,
exc.status_code,
exc.detail,
def error_response(error: ServiceError) -> JSONResponse:
return JSONResponse(
status_code=error.status_code,
content={"error": {"code": error.code, "message": error.message}},
)
return JSONResponse(
status_code=exc.status_code,
content={"detail": exc.detail},
async def service_error_handler(request: Request, exc: ServiceError) -> JSONResponse:
logger.warning(
"Service error method=%s path=%s code=%s",
request.method,
request.url.path,
exc.code,
)
return error_response(exc)
async def validation_error_handler(
request: Request, exc: RequestValidationError
) -> JSONResponse:
logger.warning("Invalid request method=%s path=%s", request.method, request.url.path)
return error_response(InvalidRequestError())
async def unexpected_error_handler(request: Request, exc: Exception) -> JSONResponse:
logger.error(
"Unexpected signing-service error method=%s path=%s type=%s",
request.method,
request.url.path,
type(exc).__name__,
)
return error_response(ServiceError())
class LoggingMiddleware(BaseHTTPMiddleware):
"""Registra request y response para trazabilidad básica."""
"""Logs only request metadata; never headers or bodies."""
async def dispatch(self, request: Request, call_next):
client_ip = request.client.host if request.client else "unknown"
method = request.method
url = request.url.path
logger.info("Request: %s %s from %s", method, url, client_ip)
logger.info("Request: %s %s from %s", request.method, request.url.path, client_ip)
response = await call_next(request)
logger.info(
"Response: %s %s returned %s to %s",
method,
url,
request.method,
request.url.path,
response.status_code,
client_ip,
)
return response
@asynccontextmanager
async def lifespan(app: FastAPI):
"""Gestiona el ciclo de vida de la aplicación."""
logger.info("Starting signing service")
yield
logger.info("Stopping signing service")
async def healthcheck() -> dict[str, str]:
"""Endpoint simple de healthcheck."""
return {"status": "ok"}
async def get_version() -> dict[str, str]:
"""Devuelve la versión publicada del servicio."""
return {"version": version}
def create_app() -> FastAPI:
app = FastAPI(lifespan=lifespan,
app = FastAPI(
lifespan=lifespan,
title="FactuGES Document Signing Service",
version=version)
app.add_exception_handler(HTTPException, http_exception_handler)
version=version,
)
app.add_exception_handler(ServiceError, service_error_handler)
app.add_exception_handler(RequestValidationError, validation_error_handler)
app.add_exception_handler(Exception, unexpected_error_handler)
app.add_middleware(LoggingMiddleware)
app.include_router(sign_router)
logger.info("API routes registered from sign_document router")
app.add_api_route("/health", healthcheck, methods=["GET"])
logger.info("Health check endpoint registered at /health")
app.add_api_route("/version", get_version, methods=["GET"])
logger.info("Version endpoint registered at /version")
return app

View File

@ -1,77 +1,53 @@
"""Opt-in, safe Infisical connectivity smoke test.
This utility never prints credentials, secret values, full secret objects, or provider
exception messages. It is not part of the automated test suite.
"""
import os
import sys
from dotenv import load_dotenv
from infisical_sdk import InfisicalSDKClient
from signing_service.application.settings.container import get_settings
load_dotenv()
def main() -> int:
settings = get_settings()
secret_name = os.environ.get("INFISICAL_SECRET_NAME", "pdf_cert_password")
secret_name = os.environ.get("INFISICAL_SECRET_NAME")
secret_path = os.environ.get("INFISICAL_SECRET_PATH", "/rodax")
missing = [k for k, v in {
"INFISICAL_SERVICE_TOKEN": settings.infisical_token_auth,
required = {
"INFISICAL_CLIENT_ID": settings.infisical_client_id,
"INFISICAL_CLIENT_SECRET": settings.infisical_client_secret,
"INFISICAL_PROJECT_ID": settings.infisical_project_id,
}.items() if not v]
"INFISICAL_ENV_SLUG": settings.infisical_env_slug,
"INFISICAL_SECRET_NAME": secret_name,
}
missing = [name for name, value in required.items() if not value]
if missing:
print(f"Missing env vars: {', '.join(missing)}", file=sys.stderr)
print("Missing required configuration", file=sys.stderr)
return 2
print("== Infisical smoke test ==")
print(f"host=https://eu.infisical.com")
print(f"client_id={settings.infisical_client_id}")
print(f"client_secret={settings.infisical_client_secret}")
print(f"project_id={settings.infisical_project_id}")
print(f"environment={settings.infisical_env_slug}")
print(f"secret_path={secret_path}")
print(f"secret_name={secret_name}")
client = InfisicalSDKClient(host="https://eu.infisical.com", token=None)
try:
# 1) Login
#client.auth.token_auth.login(
# token=settings.infisical_token_auth
#)
client = InfisicalSDKClient(host=settings.infisical_host, token=None)
client.auth.universal_auth.login(
client_id=settings.infisical_client_id,
client_secret=settings.infisical_client_secret
client_secret=settings.infisical_client_secret,
)
print("OK: token_auth.login")
# 2) Read one secret (the real test)
secret = client.secrets.get_secret_by_name(
client.secrets.get_secret_by_name(
secret_name=secret_name,
project_id=settings.infisical_project_id,
environment_slug=settings.infisical_env_slug,
secret_path=secret_path,
view_secret_value=True,
)
print(secret)
value = getattr(secret, "secretValue", None)
if not value:
print("ERROR: Secret read returned empty value", file=sys.stderr)
return 3
print("OK: get_secret_by_name")
print(f"OK: secret_value_length={len(value)}")
print("Infisical connection and secret retrieval succeeded")
return 0
except Exception as e:
# Print the most useful info without dumping secrets
print("ERROR: Infisical smoke test failed", file=sys.stderr)
print(f"{type(e).__name__}: {e}", file=sys.stderr)
except Exception as exc: # noqa: BLE001 - SDK failure types are not stable
print(f"Infisical smoke test failed ({type(exc).__name__})", file=sys.stderr)
return 1

View File

@ -0,0 +1,39 @@
import asyncio
from types import SimpleNamespace
import pytest
from signing_service.api import authentication
from signing_service.application.errors import (
AuthenticationRequiredError,
InvalidCredentialsError,
)
class Credentials:
def __init__(self, value: str, scheme: str = "Bearer") -> None:
self.credentials = value
self.scheme = scheme
def test_missing_credential_is_rejected() -> None:
with pytest.raises(AuthenticationRequiredError):
asyncio.run(authentication.require_service_authentication(None))
def test_invalid_credential_is_rejected(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(
authentication, "get_settings", lambda: SimpleNamespace(service_api_key="valid")
)
with pytest.raises(InvalidCredentialsError):
asyncio.run(authentication.require_service_authentication(Credentials("wrong")))
def test_valid_credential_is_accepted(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(
authentication, "get_settings", lambda: SimpleNamespace(service_api_key="valid")
)
assert (
asyncio.run(authentication.require_service_authentication(Credentials("valid")))
is None
)

101
tests/api/test_http.py Normal file
View File

@ -0,0 +1,101 @@
import pytest
from fastapi.testclient import TestClient
@pytest.fixture
def client(monkeypatch: pytest.MonkeyPatch) -> TestClient:
monkeypatch.setenv("APP_ENV", "local")
monkeypatch.setenv("SECRET_PROVIDER", "fake")
monkeypatch.setenv("CERT_SECRET_NAME", "pdf-signing-cert")
monkeypatch.setenv("CERT_PASSWORD_SECRET_NAME", "pdf-signing-cert-password")
monkeypatch.setenv("SERVICE_API_KEY", "test-service-key")
monkeypatch.setenv("PDF_MAX_SIZE_BYTES", "1024")
from signing_service.application.settings.container import get_settings
get_settings.cache_clear()
from signing_service.main import create_app
return TestClient(create_app(), raise_server_exceptions=False)
def form(pdf: bytes = b"%PDF-1.4 test") -> dict[str, object]:
return {
"files": {"file": ("document.pdf", pdf, "application/pdf")},
"data": {
"certificate_secret_name": "legacy-cert",
"certificate_password_secret_name": "legacy-password",
"company_slug": "legacy-company",
},
}
def test_health_is_public(client: TestClient) -> None:
response = client.get("/health")
assert response.status_code == 200
assert response.json() == {"status": "ok"}
def test_version_is_public(client: TestClient) -> None:
assert client.get("/version").status_code == 200
def test_sign_rejects_missing_authentication(client: TestClient) -> None:
response = client.post("/documents/sign", **form())
assert response.status_code == 401
assert response.json()["error"]["code"] == "AUTHENTICATION_REQUIRED"
def test_sign_rejects_invalid_authentication(client: TestClient) -> None:
response = client.post(
"/documents/sign", headers={"Authorization": "Bearer invalid"}, **form()
)
assert response.status_code == 401
assert response.json()["error"]["code"] == "INVALID_CREDENTIALS"
def test_sign_with_valid_authentication_reaches_signing_flow(
client: TestClient,
) -> None:
response = client.post(
"/documents/sign",
headers={"Authorization": "Bearer test-service-key"},
**form(),
)
assert response.status_code == 200
assert response.headers["content-type"] == "application/pdf"
assert response.content.endswith(b"-- Signed by fake signer")
def test_invalid_pdf_is_mapped_without_internal_detail(
client: TestClient, monkeypatch: pytest.MonkeyPatch
) -> None:
from signing_service.api.routes import sign_document
from signing_service.infrastructure.pdf.pdf_validator_error import (
PDFValidationError,
)
class InvalidPdfSigner:
async def sign(self, **kwargs: object) -> bytes:
raise PDFValidationError("PFX parsing detail")
monkeypatch.setattr(sign_document, "create_pdf_signer", lambda: InvalidPdfSigner())
response = client.post(
"/documents/sign",
headers={"Authorization": "Bearer test-service-key"},
**form(b"not a PDF"),
)
assert response.status_code == 400
assert response.json() == {
"error": {"code": "INVALID_PDF", "message": "Invalid PDF document"}
}
def test_pdf_size_limit_is_controlled(client: TestClient) -> None:
response = client.post(
"/documents/sign",
headers={"Authorization": "Bearer test-service-key"},
**form(b"x" * 1025),
)
assert response.status_code == 413
assert response.json()["error"]["code"] == "INVALID_REQUEST"

View File

@ -1,59 +1,96 @@
import asyncio
import pytest
from signing_service.application.use_cases.sign_document import (
SignDocumentUseCase,
)
from signing_service.application.use_cases.sign_document_dto import (
SignDocumentCommand,
from signing_service.application.errors import (
InvalidPdfError,
SecretProviderError,
ServiceError,
)
from signing_service.application.use_cases.sign_document import SignDocumentUseCase
from signing_service.application.use_cases.sign_document_dto import SignDocumentCommand
from signing_service.infrastructure.pdf.fake_pdf_signer import FakePDFSigner
from signing_service.infrastructure.secrets.fake_secret_manager import (
FakeSecretManager,
)
from signing_service.infrastructure.pdf.pdf_validator_error import PDFValidationError
from signing_service.infrastructure.secrets.fake_secret_manager import FakeSecretManager
CERT_NAME = "configured-cert"
PASSWORD_NAME = "configured-password"
def test_sign_document_success():
# GIVEN: dependencias fake
fake_secrets = {
"pdf-signing-cert": "FAKE_CERTIFICATE_DATA"
}
secret_manager = FakeSecretManager(fake_secrets)
pdf_signer = FakePDFSigner()
def command() -> SignDocumentCommand:
return SignDocumentCommand(
pdf_bytes=b"%PDF-1.4 fake pdf content",
certificate_secret_name="legacy-cert-name",
certificate_password_secret_name="legacy-password-name",
company_slug="legacy-company",
)
def test_sign_document_success() -> None:
use_case = SignDocumentUseCase(
secret_manager=secret_manager,
pdf_signer=pdf_signer,
secret_manager=FakeSecretManager(
{CERT_NAME: "certificate", PASSWORD_NAME: "password"}
),
pdf_signer=FakePDFSigner(),
cert_secret_name=CERT_NAME,
cert_password_secret_name=PASSWORD_NAME,
)
original_pdf = b"%PDF-1.4 fake pdf content"
result = asyncio.run(use_case.execute(command()))
command = SignDocumentCommand(
pdf_bytes=original_pdf,
certificate_secret_name="pdf-signing-cert",
certificate_password_secret_name="pdf-signing-cert-password",
)
# WHEN: ejecutamos el caso de uso
result = use_case.execute(command)
# THEN: el PDF está "firmado"
assert result.signed_pdf_bytes.startswith(original_pdf)
assert b"FAKE_CERTIFICATE_DATA" in result.signed_pdf_bytes
assert result.signed_pdf_bytes.startswith(command().pdf_bytes)
assert result.signed_pdf_bytes.endswith(b"-- Signed by fake signer")
assert b"certificate" not in result.signed_pdf_bytes
assert b"password" not in result.signed_pdf_bytes
def test_sign_document_secret_not_found():
secret_manager = FakeSecretManager(secrets={})
pdf_signer = FakePDFSigner()
def test_sign_document_maps_secret_manager_errors() -> None:
use_case = SignDocumentUseCase(
secret_manager=secret_manager,
pdf_signer=pdf_signer,
secret_manager=FakeSecretManager({}),
pdf_signer=FakePDFSigner(),
cert_secret_name=CERT_NAME,
cert_password_secret_name=PASSWORD_NAME,
)
command = SignDocumentCommand(
pdf_bytes=b"fake pdf",
certificate_secret_name="missing-secret",
with pytest.raises(SecretProviderError):
asyncio.run(use_case.execute(command()))
class FailingSigner:
async def sign(self, **kwargs: object) -> bytes:
raise RuntimeError("PFX password was rejected")
class InvalidPdfSigner:
async def sign(self, **kwargs: object) -> bytes:
raise PDFValidationError("internal parsing detail")
def test_sign_document_maps_signer_errors() -> None:
use_case = SignDocumentUseCase(
secret_manager=FakeSecretManager(
{CERT_NAME: "certificate", PASSWORD_NAME: "password"}
),
pdf_signer=FailingSigner(), # type: ignore[arg-type]
cert_secret_name=CERT_NAME,
cert_password_secret_name=PASSWORD_NAME,
)
with pytest.raises(ValueError):
use_case.execute(command)
with pytest.raises(ServiceError) as caught:
asyncio.run(use_case.execute(command()))
assert caught.value.code == "SIGNING_FAILED"
def test_sign_document_maps_invalid_pdf_errors() -> None:
use_case = SignDocumentUseCase(
secret_manager=FakeSecretManager(
{CERT_NAME: "certificate", PASSWORD_NAME: "password"}
),
pdf_signer=InvalidPdfSigner(), # type: ignore[arg-type]
cert_secret_name=CERT_NAME,
cert_password_secret_name=PASSWORD_NAME,
)
with pytest.raises(InvalidPdfError):
asyncio.run(use_case.execute(command()))