office_translator/_bmad-output/implementation-artifacts/3-7-webhook-specification-url.md

Story 3.7: Webhook - Spécification URL

Status: ready-for-dev

Story

En tant qu'utilisateur, Je veux pouvoir spécifier une URL de webhook dans ma requête de traduction, de sorte que je puisse être notifié automatiquement quand la traduction est terminée.

Acceptance Criteria

1. Paramètre webhook_url accepté: POST /api/v1/translate accepte un paramètre webhook_url optionnel. (FR36, FR65)
2. Validation format URL: Si webhook_url est fourni, il doit être une URL HTTP/HTTPS valide.
3. Erreur INVALID_WEBHOOK_URL: Si le format est invalide, retourne 400 avec error code INVALID_WEBHOOK_URL.
4. Stockage avec le job: webhook_url est stocké avec le job de traduction pour utilisation ultérieure.
5. Paramètre optionnel: webhook_url est optionnel - la traduction fonctionne sans webhook. (FR65)
6. Documentation OpenAPI: Le paramètre webhook_url est documenté avec description et exemple.

Tasks / Subtasks

• Task 1: Améliorer la validation du webhook_url (AC: #2, #3)

  • 1.1 Créer une fonction validate_webhook_url() dédiée dans middleware/validation.py
  • 1.2 Valider le format URL (http/https, domaine valide, pas d'IP privée)
  • 1.3 Retourner l'erreur INVALID_WEBHOOK_URL avec détails appropriés
  • 1.4 Ajouter des tests unitaires pour la validation

• Task 2: Mettre à jour le endpoint translate (AC: #1, #4)

  • 2.1 Vérifier que webhook_url est déjà accepté comme paramètre Form
  • 2.2 Intégrer la nouvelle validation dans le flux du endpoint
  • 2.3 S'assurer que webhook_url est stocké dans _translation_jobs[job_id]
  • 2.4 Vérifier que le code existant de notification webhook (Story 3.8) utilise bien cette valeur

• Task 3: Ajouter le code d'erreur INVALID_WEBHOOK_URL (AC: #3)

  • 3.1 Vérifier que INVALID_WEBHOOK_URL existe déjà dans schemas/errors.py
  • 3.2 Ajouter un exemple d'erreur dans ERROR_EXAMPLES si manquant
  • 3.3 Documenter le code d'erreur dans la docstring du endpoint

• Task 4: Documenter le paramètre dans OpenAPI (AC: #6)

  • 4.1 Ajouter une description complète au paramètre webhook_url dans le Form
  • 4.2 Ajouter un exemple de requête avec webhook_url dans la documentation
  • 4.3 Documenter le comportement Fire & Forget (Story 3.8)

• Task 5: Tests d'intégration (AC: Tous)

  • 5.1 Tester une requête avec webhook_url valide
  • 5.2 Tester une requête avec webhook_url invalide (mauvais format)
  • 5.3 Tester une requête avec webhook_url invalide (URL non HTTP/HTTPS)
  • 5.4 Tester une requête sans webhook_url (devrait réussir)
  • 5.5 Vérifier que le webhook_url est bien stocké dans le job

Dev Notes

État Actuel de l'Implémentation

CE QUI EXISTE DÉJÀ dans routes/translate_routes.py:

# Ligne ~350 - Paramètre déjà accepté
webhook_url: Optional[str] = Form(None, description="Webhook URL for notification"),

# Ligne ~400 - Validation basique existante
if webhook_url:
    if not re.match(r"^https?://", webhook_url, re.IGNORECASE):
        raise TranslateEndpointError(
            code="INVALID_FORMAT",  # ⚠️ Devrait être INVALID_WEBHOOK_URL
            message="URL webhook invalide. Doit commencer par http:// ou https://",
            details={"field": "webhook_url"},
        )

# Ligne ~470 - Stockage dans le job
"webhook_url": webhook_url,

# Ligne ~560-575 - Notification webhook existante (Story 3.8)
if webhook_url:
    try:
        async with httpx.AsyncClient(timeout=10) as client:
            await client.post(
                webhook_url,
                json={
                    "translation_id": job_id,
                    "status": job["status"],
                    "timestamp": datetime.now(timezone.utc).isoformat(),
                    "file_name": job.get("file_name"),
                    "error_message": job.get("error_message"),
                },
            )
    except Exception as e:
        logger.warning(f"Job {job_id}: Webhook notification failed - {e}")

CE QUI MANQUE:

• ❌ Code d'erreur INVALID_WEBHOOK_URL au lieu de INVALID_FORMAT
• ❌ Validation plus robuste (domaine valide, pas d'IP privée pour sécurité)
• ❌ Documentation OpenAPI complète pour le paramètre
• ❌ Tests unitaires dédiés à la validation webhook

Architecture Patterns

Validation Pattern (depuis middleware/validation.py):

class ValidationError(Exception):
    """Exception for validation errors with structured error codes."""
    def __init__(self, code: str, message: str, details: Optional[dict] = None):
        self.code = code
        self.message = message
        self.details = details or {}

class WebhookURLValidator:
    """Validator for webhook URLs."""
    
    def __init__(self, allowed_schemes: tuple = ("http", "https")):
        self.allowed_schemes = allowed_schemes
    
    def validate(self, url: str) -> ValidationResult:
        """Validate webhook URL format and security."""
        # Implementation
        pass

Error Response Format (depuis schemas/errors.py):

class ErrorCode(str, Enum):
    # ...
    INVALID_WEBHOOK_URL = "INVALID_WEBHOOK_URL"
    # ...

ERROR_EXAMPLES = {
    # ...
    "INVALID_WEBHOOK_URL": {
        "summary": "URL webhook invalide",
        "value": {
            "error": "INVALID_WEBHOOK_URL",
            "message": "L'URL du webhook doit être une URL HTTP/HTTPS valide.",
            "details": {
                "field": "webhook_url",
                "hint": "L'URL doit commencer par http:// ou https://"
            }
        }
    }
}

Sécurité - Validation Webhook URL

Risques de sécurité à considérer:

1. SSRF (Server-Side Request Forgery):

   • Ne pas permettre les URLs vers des IPs privées (10.x, 172.16-31.x, 192.168.x)
   • Ne pas permettre localhost (127.0.0.1, ::1)
   • Ne pas permettre les URLs avec credentials intégrés

2. Validation recommandée:

import ipaddress
from urllib.parse import urlparse

def validate_webhook_url(url: str) -> tuple[bool, Optional[str]]:
    """
    Validate webhook URL for security.
    
    Returns: (is_valid, error_message)
    """
    try:
        parsed = urlparse(url)
        
        # Check scheme
        if parsed.scheme not in ("http", "https"):
            return False, "L'URL doit utiliser http:// ou https://"
        
        # Check for credentials in URL
        if parsed.username or parsed.password:
            return False, "L'URL ne doit pas contenir de credentials"
        
        # Resolve hostname and check for private IPs
        import socket
        hostname = parsed.hostname
        if hostname:
            # Block localhost
            if hostname.lower() in ("localhost", "127.0.0.1", "::1"):
                return False, "Les URLs localhost ne sont pas autorisées"
            
            # Resolve and check IP
            try:
                ip_str = socket.gethostbyname(hostname)
                ip = ipaddress.ip_address(ip_str)
                if ip.is_private or ip.is_loopback or ip.is_link_local:
                    return False, "Les adresses IP privées ne sont pas autorisées"
            except socket.gaierror:
                pass  # DNS resolution failed, let it through (will fail at webhook time)
        
        return True, None
        
    except Exception as e:
        return False, f"Format d'URL invalide: {str(e)}"

Project Structure Notes

• Fichier principal: routes/translate_routes.py (modifier validation existante)
• Validation: middleware/validation.py (ajouter WebhookURLValidator)
• Erreurs: schemas/errors.py (vérifier INVALID_WEBHOOK_URL)
• Tests: tests/test_webhook_validation.py (nouveau fichier)

Références

• [Source: _bmad-output/planning-artifacts/epics.md#Story 3.7]
• [Source: _bmad-output/planning-artifacts/architecture.md#API & Communication Patterns]
• [Source: _bmad-output/planning-artifacts/architecture.md#Error Handling]
• [Source: FR36, FR65, FR66 - Webhook requirements]
• [Source: routes/translate_routes.py - Implémentation actuelle]

Intelligence des Stories Précédentes (Epic 3)

Story 3.1 (API Key Generation) - Enseignements

1. Routeur api_key_routes.py utilise prefix="/api/v1/api-keys" ✅
2. Pattern de réponse {data: {...}, meta: {...}} établi
3. Tests complets avec fixtures dans tests/conftest.py

Story 3.2 (API Key Revocation User) - Enseignements

1. Soft delete avec is_active=False
2. Fonction get_user_by_api_key dans services/auth_service.py
3. Format d'erreur structuré {error, message, details?}

Story 3.3 (Admin API Key Revocation) - Enseignements

1. Routes admin dans main.py - à migrer vers routes/admin_routes.py
2. Dépendance require_admin pour l'authentification admin
3. Audit logging avec logger.info()

Story 3.4 (API Auth X-API-Key) - Enseignements

1. Coexistence JWT + API Key dans get_authenticated_user
2. Middleware d'auth dans routes/translate_routes.py
3. Priorité API key sur JWT si les deux présents

Story 3.5 (API Versioning) - Enseignements

1. Tous les endpoints sont maintenant sous /api/v1/ ✅
2. Exceptions documentées: /health, /ready, /docs, /redoc
3. Routeur principal routes/api_v1_router.py

Story 3.6 (Documentation OpenAPI) - Enseignements

1. Schémas Pydantic dans schemas/ pour documentation ✅
2. Codes d'erreur documentés dans schemas/errors.py avec ERROR_EXAMPLES ✅
3. INVALID_WEBHOOK_URL existe déjà dans ErrorCode enum ✅
4. Descriptions complètes dans les docstrings des endpoints

Contexte Métier

Epic 3: API & Automation (Pro)

Cette story est la septième de l'Epic 3 qui permet aux utilisateurs Pro d'automatiser les traductions:

1. API Keys - Génération (Story 3.1 ✅)
2. API Keys - Révocation User (Story 3.2 ✅)
3. API Keys - Révocation Admin (Story 3.3 ✅)
4. Authentification X-API-Key (Story 3.4 ✅)
5. API Versioning (Story 3.5 ✅)
6. Documentation OpenAPI (Story 3.6 ✅)
7. Webhook - Spécification URL (cette story)
8. Webhook - Envoi POST Fire & Forget (Story 3.8 - backlog)
9. Glossaires (Stories 3.9-3.10 - backlog)
10. Custom Prompts (Stories 3.11-3.12 - backlog)

Valeur Business

Le paramètre webhook_url est critique pour:

• Automation: Thomas (Pro user) peut enchaîner les traductions dans ses workflows n8n
• Intégration CI/CD: Notification automatique quand une traduction termine
• Architecture événementielle: Base pour les futures fonctionnalités temps réel
• Expérience Pro: Justification de l'abonnement Pro

Dépendances

• Stories 3.1-3.6 (prérequis): API versionnée, authentifiée, documentée ✅
• Story 3.8 (suivante): Envoi du webhook - utilisera le webhook_url stocké
• Stories futures: Glossaires et Custom Prompts pourront aussi utiliser les webhooks

Guardrails Développeur

❌ À NE PAS FAIRE

1. NE PAS utiliser le code d'erreur INVALID_FORMAT pour les webhooks - utiliser INVALID_WEBHOOK_URL
2. NE PAS permettre les URLs vers des IPs privées (risque SSRF)
3. NE PAS permettre les URLs avec credentials intégrés (http://user:pass@...)
4. NE PAS bloquer la traduction si le webhook échoue (Fire & Forget)
5. NE PAS stocker le webhook_url en base de données - seulement en mémoire avec le job
6. NE PAS oublier de documenter le paramètre dans OpenAPI

✅ À FAIRE

1. TOUJOURS valider le format URL (http/https, domaine valide)
2. TOUJOURS utiliser le code d'erreur INVALID_WEBHOOK_URL pour les erreurs de validation
3. TOUJOURS rendre le paramètre optionnel (la traduction fonctionne sans webhook)
4. TOUJOURS stocker le webhook_url avec le job pour la Story 3.8
5. CRÉER une fonction de validation dédiée dans middleware/validation.py
6. AJOUTER des tests unitaires pour la validation webhook
7. DOCUMENTER le paramètre avec description et exemple dans OpenAPI

Code Suggéré

Fichier middleware/validation.py - Ajouter WebhookURLValidator

import re
import ipaddress
import socket
from urllib.parse import urlparse
from typing import Optional, Tuple


class WebhookURLValidator:
    """
    Validator for webhook URLs with security checks.
    
    Prevents SSRF attacks by blocking private IPs and localhost.
    """
    
    # Allowed URL schemes
    ALLOWED_SCHEMES = ("http", "https")
    
    # Blocked hostnames
    BLOCKED_HOSTNAMES = {"localhost", "127.0.0.1", "::1", "0.0.0.0"}
    
    def __init__(
        self,
        allowed_schemes: Tuple[str, ...] = ALLOWED_SCHEMES,
        block_private_ips: bool = True
    ):
        self.allowed_schemes = allowed_schemes
        self.block_private_ips = block_private_ips
    
    def validate(self, url: str) -> Tuple[bool, Optional[str], Optional[dict]]:
        """
        Validate webhook URL format and security.
        
        Args:
            url: The webhook URL to validate
            
        Returns:
            Tuple of (is_valid, error_message, details)
        """
        if not url:
            return True, None, None
        
        try:
            parsed = urlparse(url)
            
            # Check scheme
            if parsed.scheme.lower() not in self.allowed_schemes:
                return False, (
                    f"L'URL doit utiliser {' ou '.join(self.allowed_schemes)}",
                    {
                        "field": "webhook_url",
                        "allowed_schemes": list(self.allowed_schemes),
                        "detected_scheme": parsed.scheme or "none"
                    }
                )
            
            # Check for credentials in URL
            if parsed.username or parsed.password:
                return False, (
                    "L'URL ne doit pas contenir de credentials",
                    {"field": "webhook_url", "reason": "credentials_in_url"}
                )
            
            # Check hostname
            hostname = parsed.hostname
            if not hostname:
                return False, (
                    "URL invalide: hostname manquant",
                    {"field": "webhook_url", "reason": "missing_hostname"}
                )
            
            # Block localhost and common local addresses
            if hostname.lower() in self.BLOCKED_HOSTNAMES:
                return False, (
                    "Les URLs localhost ne sont pas autorisées",
                    {"field": "webhook_url", "reason": "localhost_blocked"}
                )
            
            # Check for private IPs (SSRF protection)
            if self.block_private_ips:
                try:
                    # Try to parse as IP directly
                    try:
                        ip = ipaddress.ip_address(hostname)
                        if self._is_blocked_ip(ip):
                            return False, (
                                "Les adresses IP privées ne sont pas autorisées",
                                {"field": "webhook_url", "reason": "private_ip_blocked"}
                            )
                    except ValueError:
                        # Not an IP, try DNS resolution
                        ip_str = socket.gethostbyname(hostname)
                        ip = ipaddress.ip_address(ip_str)
                        if self._is_blocked_ip(ip):
                            return False, (
                                "Les adresses IP privées ne sont pas autorisées",
                                {"field": "webhook_url", "reason": "private_ip_blocked"}
                            )
                except socket.gaierror:
                    # DNS resolution failed - let it through
                    # Will fail at webhook send time
                    pass
                except Exception:
                    pass
            
            return True, None, None
            
        except Exception as e:
            return False, (
                f"Format d'URL invalide: {str(e)}",
                {"field": "webhook_url", "error": str(e)}
            )
    
    def _is_blocked_ip(self, ip: ipaddress.IPv4Address | ipaddress.IPv6Address) -> bool:
        """Check if IP is private, loopback, or link-local."""
        return (
            ip.is_private or
            ip.is_loopback or
            ip.is_link_local or
            ip.is_reserved or
            ip.is_multicast
        )


# Singleton instance
webhook_validator = WebhookURLValidator()

Modification de routes/translate_routes.py

# Ajouter l'import
from middleware.validation import webhook_validator

# Remplacer la validation existante (ligne ~400)
# AVANT:
if webhook_url:
    if not re.match(r"^https?://", webhook_url, re.IGNORECASE):
        raise TranslateEndpointError(
            code="INVALID_FORMAT",
            message="URL webhook invalide. Doit commencer par http:// ou https://",
            details={"field": "webhook_url"},
        )

# APRÈS:
if webhook_url:
    is_valid, error_msg, error_details = webhook_validator.validate(webhook_url)
    if not is_valid:
        raise TranslateEndpointError(
            code="INVALID_WEBHOOK_URL",
            message=error_msg,
            details=error_details,
        )

Ajout dans schemas/errors.py - ERROR_EXAMPLES

ERROR_EXAMPLES = {
    # ... existing examples ...
    
    "INVALID_WEBHOOK_URL": {
        "summary": "URL webhook invalide",
        "value": {
            "error": "INVALID_WEBHOOK_URL",
            "message": "L'URL du webhook doit être une URL HTTP/HTTPS valide.",
            "details": {
                "field": "webhook_url",
                "allowed_schemes": ["http", "https"],
                "hint": "L'URL doit commencer par http:// ou https://"
            }
        }
    },
    "WEBHOOK_PRIVATE_IP": {
        "summary": "IP privée non autorisée",
        "value": {
            "error": "INVALID_WEBHOOK_URL",
            "message": "Les adresses IP privées ne sont pas autorisées.",
            "details": {
                "field": "webhook_url",
                "reason": "private_ip_blocked"
            }
        }
    },
    "WEBHOOK_LOCALHOST": {
        "summary": "Localhost non autorisé",
        "value": {
            "error": "INVALID_WEBHOOK_URL",
            "message": "Les URLs localhost ne sont pas autorisées.",
            "details": {
                "field": "webhook_url",
                "reason": "localhost_blocked"
            }
        }
    }
}

Fichier de tests tests/test_webhook_validation.py

"""
Tests for webhook URL validation.
Story 3.7: Webhook - Spécification URL
"""

import pytest
from middleware.validation import webhook_validator, WebhookURLValidator


class TestWebhookURLValidator:
    """Tests for WebhookURLValidator class."""
    
    def test_valid_https_url(self):
        """Valid HTTPS URL should pass."""
        url = "https://example.com/webhook"
        is_valid, error, details = webhook_validator.validate(url)
        assert is_valid is True
        assert error is None
        assert details is None
    
    def test_valid_http_url(self):
        """Valid HTTP URL should pass."""
        url = "http://example.com/webhook"
        is_valid, error, details = webhook_validator.validate(url)
        assert is_valid is True
        assert error is None
    
    def test_invalid_scheme_ftp(self):
        """FTP URL should be rejected."""
        url = "ftp://example.com/webhook"
        is_valid, error, details = webhook_validator.validate(url)
        assert is_valid is False
        assert "http" in error.lower()
    
    def test_invalid_scheme_no_scheme(self):
        """URL without scheme should be rejected."""
        url = "example.com/webhook"
        is_valid, error, details = webhook_validator.validate(url)
        assert is_valid is False
    
    def test_localhost_blocked(self):
        """Localhost should be blocked."""
        urls = [
            "http://localhost/webhook",
            "http://127.0.0.1/webhook",
            "http://[::1]/webhook",
            "http://0.0.0.0/webhook",
        ]
        for url in urls:
            is_valid, error, details = webhook_validator.validate(url)
            assert is_valid is False, f"URL {url} should be blocked"
            assert "localhost" in error.lower() or "priv" in error.lower()
    
    def test_credentials_in_url_blocked(self):
        """URLs with credentials should be blocked."""
        url = "https://user:password@example.com/webhook"
        is_valid, error, details = webhook_validator.validate(url)
        assert is_valid is False
        assert "credentials" in error.lower()
    
    def test_empty_url_valid(self):
        """Empty URL should be valid (optional parameter)."""
        is_valid, error, details = webhook_validator.validate("")
        assert is_valid is True
    
    def test_none_url_valid(self):
        """None URL should be valid (optional parameter)."""
        is_valid, error, details = webhook_validator.validate(None)
        # Note: The validator should handle None gracefully
        # Implementation may vary
    
    def test_url_with_port(self):
        """URL with port should be valid."""
        url = "https://example.com:8080/webhook"
        is_valid, error, details = webhook_validator.validate(url)
        assert is_valid is True
    
    def test_url_with_query_params(self):
        """URL with query parameters should be valid."""
        url = "https://example.com/webhook?token=abc123&source=api"
        is_valid, error, details = webhook_validator.validate(url)
        assert is_valid is True
    
    def test_url_with_path(self):
        """URL with path should be valid."""
        url = "https://example.com/api/v1/notifications/translation-complete"
        is_valid, error, details = webhook_validator.validate(url)
        assert is_valid is True


class TestWebhookURLIntegration:
    """Integration tests for webhook URL in translate endpoint."""
    
    @pytest.mark.asyncio
    async def test_translate_with_valid_webhook(self, client, auth_headers, sample_file):
        """Translation with valid webhook_url should succeed."""
        # Implementation
        pass
    
    @pytest.mark.asyncio
    async def test_translate_with_invalid_webhook(self, client, auth_headers, sample_file):
        """Translation with invalid webhook_url should return 400."""
        # Implementation
        pass
    
    @pytest.mark.asyncio
    async def test_translate_without_webhook(self, client, auth_headers, sample_file):
        """Translation without webhook_url should succeed."""
        # Implementation
        pass

Dev Agent Record

Agent Model Used

{{agent_model_name_version}}

Debug Log References

(À remplir lors de l'implémentation)

Completion Notes List

• ✅ Analyse exhaustive du contexte terminée
• ✅ Code existant identifié dans routes/translate_routes.py
• ✅ Validation basique existante à améliorer
• ✅ Code d'erreur INVALID_WEBHOOK_URL déjà présent dans schemas/errors.py
• ✅ Tests à créer dans tests/test_webhook_validation.py

File List

• middleware/validation.py - À MODIFIER - Ajouter WebhookURLValidator
• routes/translate_routes.py - À MODIFIER - Utiliser nouvelle validation
• schemas/errors.py - À VÉRIFIER - ERROR_EXAMPLES pour INVALID_WEBHOOK_URL
• tests/test_webhook_validation.py - À CRÉER - Tests unitaires validation webhook

Change Log

• 2026-02-22: Story créée avec contexte complet et analyse exhaustive

Checklist de Validation

Avant de marquer cette story comme terminée, vérifier:

• Le paramètre webhook_url est accepté dans POST /api/v1/translate
• La validation utilise le code d'erreur INVALID_WEBHOOK_URL (pas INVALID_FORMAT)
• Les URLs localhost et IPs privées sont bloquées (sécurité SSRF)
• Les URLs avec credentials sont bloquées
• Le paramètre est optionnel (la traduction fonctionne sans)
• Le webhook_url est stocké avec le job pour la Story 3.8
• Le paramètre est documenté dans OpenAPI
• Les tests unitaires passent
• Les tests d'intégration passent