haproxy-openmanager/backend/models/ssl.py

from pydantic import BaseModel, field_validator, model_validator
from typing import Optional, List, Dict, Any
from datetime import datetime
import logging

logger = logging.getLogger(__name__)

class SSLCertificateCreate(BaseModel):
    name: str
    certificate_content: str  # PEM format certificate
    private_key_content: Optional[str] = None  # PEM format private key (optional for server SSL)
    chain_content: Optional[str] = None  # PEM format certificate chain (optional)
    cluster_ids: Optional[List[int]] = None  # List of cluster IDs for multi-cluster support
    is_global: bool = False  # True for global SSL certificates
    usage_type: str = "frontend"  # "frontend" or "server" - determines if private key is required
    
    @field_validator('usage_type')
    @classmethod
    def validate_usage_type(cls, v):
        if v not in ['frontend', 'server']:
            raise ValueError('usage_type must be either "frontend" or "server"')
        return v

    @field_validator('name')
    @classmethod
    def validate_name_no_path_traversal(cls, v):
        """Bulgu #21 (round-11 audit): SSL certificate `name` is interpolated
        into the on-disk certificate path:

            cert_path = f"/etc/ssl/haproxy/{ssl_cert['name']}.pem"

        and emitted into the rendered HAProxy config. The agent then
        runs `mv "$temp_cert_file" "$cert_file_path"` as root, which
        means a name like '../../tmp/evil' resolves to '/tmp/evil.pem'
        and lets an operator with ssl.create permission overwrite
        arbitrary `*.pem` files on the agent host. The trailing `.pem`
        suffix mitigates common exploit paths (cron.d, profile.d,
        authorized_keys) but is defense-only — the right fix is to
        constrain `name` at the API boundary. Mirror SSLChoice's
        constraints so the wizard and direct API give the same answer.
        """
        if v is None:
            return v
        stripped = v.strip()
        if not stripped:
            raise ValueError('SSL certificate name must not be empty')
        if stripped != v:
            raise ValueError('SSL certificate name must not contain leading/trailing whitespace')
        if len(stripped) > 200:
            raise ValueError('SSL certificate name must be 200 characters or fewer')
        import re as _re
        if not _re.match(r'^[A-Za-z0-9_.-]+$', stripped):
            raise ValueError(
                f'SSL certificate name={v!r} contains forbidden characters '
                '— only letters, digits, underscore, hyphen, and dot are '
                'allowed (the name is used as a filename component under '
                '/etc/ssl/haproxy/).'
            )
        if '..' in stripped:
            raise ValueError(f'SSL certificate name={v!r} must not contain ".." (path traversal)')
        if stripped.startswith('.'):
            raise ValueError(f'SSL certificate name={v!r} must not start with "." (hidden filename)')
        if stripped.startswith('-'):
            raise ValueError(f'SSL certificate name={v!r} must not start with "-" (CLI flag confusion)')
        return stripped

    @field_validator('certificate_content')
    @classmethod
    def validate_certificate(cls, v):
        if not v or not v.strip():
            raise ValueError('Certificate content is required')
        
        # Basic PEM format check
        v = v.strip()
        if '-----BEGIN CERTIFICATE-----' not in v or '-----END CERTIFICATE-----' not in v:
            raise ValueError('Certificate must be in PEM format')
        
        return v
    
    @model_validator(mode='after')
    def validate_private_key_based_on_usage(self):
        """Private key is required for frontend SSL, optional for server SSL"""
        usage_type = self.usage_type
        private_key = self.private_key_content
        
        if usage_type == 'frontend':
            # Frontend SSL requires private key
            if not private_key or not private_key.strip():
                raise ValueError('Private key is required for frontend SSL certificates')
            
            # Basic PEM format check
            private_key = private_key.strip()
            if '-----BEGIN' not in private_key or '-----END' not in private_key:
                raise ValueError('Private key must be in PEM format')
        elif usage_type == 'server':
            # Server SSL - private key is optional
            if private_key and private_key.strip():
                # If provided, validate format
                private_key = private_key.strip()
                if '-----BEGIN' not in private_key or '-----END' not in private_key:
                    raise ValueError('Private key must be in PEM format')
        
        return self
    
    @field_validator('chain_content')
    @classmethod
    def validate_chain(cls, v):
        if v and v.strip():
            # Basic PEM format check for chain
            if '-----BEGIN CERTIFICATE-----' not in v or '-----END CERTIFICATE-----' not in v:
                raise ValueError('Certificate chain must be in PEM format')
        
        return v

class SSLCertificateUpdate(BaseModel):
    name: Optional[str] = None
    certificate_content: Optional[str] = None
    private_key_content: Optional[str] = None
    chain_content: Optional[str] = None
    cluster_id: Optional[int] = None
    usage_type: Optional[str] = None  # "frontend" or "server"
    
    @field_validator('usage_type')
    @classmethod
    def validate_usage_type(cls, v):
        if v is not None and v not in ['frontend', 'server']:
            raise ValueError('usage_type must be either "frontend" or "server"')
        return v

    # Bulgu #63 (round-22 audit) — the path-traversal guard previously
    # lived here as a strict Pydantic validator (Bulgu #21, round-11).
    # On UPDATE the manual SSL UI re-sends the existing `name` along
    # with the field the operator actually edited (usage_type /
    # cluster_id / content). If the existing certificate name was
    # imported BEFORE Bulgu #21 landed (or by a different upload path)
    # and contained a now-rejected character — e.g. legacy uploads
    # with `cert (1).pem`, `*.example.com`, `client cert.pem`, or
    # `wildcard.example.com` if dot was not yet escaped — every PUT
    # would 400 even though the operator was only changing usage
    # type or attaching the cert to another cluster. That mirrors
    # the Bulgu #62 lockout on frontends.
    #
    # The guard was moved into `routers/ssl.py::_assert_safe_cert_name`
    # and is now invoked by the create + update routes:
    #   * create: strict — rejects unsafe names outright (Bulgu #21
    #     contract preserved for fresh inserts).
    #   * update: skipped if `name` is unchanged from the DB row;
    #     enforced strictly only when the operator actually renames
    #     the certificate.

class SSLCertificate(BaseModel):
    id: int
    name: str
    domain: str  # Auto-parsed from certificate
    all_domains: List[str] = []  # All domains from SAN + CN
    certificate_content: str
    private_key_content: Optional[str] = None  # Optional for server SSL
    chain_content: Optional[str] = None
    expiry_date: Optional[datetime] = None  # Auto-parsed from certificate
    issuer: Optional[str] = None  # Auto-parsed from certificate
    status: str  # 'valid', 'expiring_soon', 'expired', 'invalid'
    days_until_expiry: int = 0
    fingerprint: Optional[str] = None
    cluster_id: Optional[int] = None
    usage_type: str = "frontend"  # "frontend" or "server"
    created_at: Optional[datetime] = None
    updated_at: Optional[datetime] = None
    has_pending_config: bool = False
    
    # Certificate details for UI display
    certificate_info: Optional[Dict[str, Any]] = None

class SSLCertificateResponse(BaseModel):
    """Response model for SSL certificate API endpoints"""
    id: int
    name: str
    domain: str
    all_domains: List[str] = []
    status: str
    days_until_expiry: int = 0
    expiry_date: Optional[datetime] = None
    issuer: Optional[str] = None
    fingerprint: Optional[str] = None
    cluster_id: Optional[int] = None
    usage_type: str = "frontend"  # "frontend" or "server"
    created_at: Optional[datetime] = None
    updated_at: Optional[datetime] = None
    has_pending_config: bool = False