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

# Story 3.7: Webhook - Spécification URL

Status: ready-for-dev

<!-- Note: Validation is optional. Run validate-create-story for quality check before dev-story. -->

## 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`:

```python
# 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`):

```python
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`):

```python
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**:
```python
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

```python
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`

```python
# 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

```python
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`

```python
"""
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