feat(security): implement IP whitelist for production environments

- Add IP whitelist service with CIDR support
- Create middleware for IP validation and enforcement
- Implement admin API endpoints for whitelist management
- Add database migration for ip_whitelists table
- Configure automatic defaults for different environments
- Support expiration dates and metadata for entries
- Include comprehensive unit tests for service logic

alembic/versions/006_add_ip_whitelist_table.py

@@ -0,0 +1,70 @@
+"""add ip whitelist table
+
+Revision ID: 006
+Revises: 005
+Create Date: 2025-01-25
+"""
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+# revision identifiers
+revision = '006'
+down_revision = '005'
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    """Create ip_whitelists table."""
+    op.create_table(
+        'ip_whitelists',
+        sa.Column('id', sa.String(64), primary_key=True),
+        sa.Column('ip_address', sa.String(45), nullable=False),
+        sa.Column('description', sa.String(255), nullable=True),
+        sa.Column('environment', sa.String(20), nullable=False, server_default='production'),
+        sa.Column('active', sa.Boolean(), nullable=False, server_default='true'),
+        sa.Column('created_by', sa.String(255), nullable=False),
+        sa.Column('created_at', sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
+        sa.Column('expires_at', sa.DateTime(timezone=True), nullable=True),
+        sa.Column('metadata', postgresql.JSON(), nullable=False, server_default='{}'),
+        sa.Column('is_cidr', sa.Boolean(), nullable=False, server_default='false'),
+        sa.Column('cidr_prefix', sa.Integer(), nullable=True),
+    )
+    
+    # Create indexes
+    op.create_index(
+        'ix_ip_whitelists_ip_address',
+        'ip_whitelists',
+        ['ip_address']
+    )
+    
+    op.create_index(
+        'ix_ip_whitelists_environment',
+        'ip_whitelists',
+        ['environment']
+    )
+    
+    op.create_index(
+        'ix_ip_whitelists_active',
+        'ip_whitelists',
+        ['active']
+    )
+    
+    op.create_index(
+        'ix_ip_whitelists_expires_at',
+        'ip_whitelists',
+        ['expires_at']
+    )
+    
+    # Create unique constraint on ip + environment
+    op.create_unique_constraint(
+        'uq_ip_whitelists_ip_environment',
+        'ip_whitelists',
+        ['ip_address', 'environment']
+    )
+
+
+def downgrade() -> None:
+    """Drop ip_whitelists table."""
+    op.drop_table('ip_whitelists')

src/api/app.py

@@ -27,6 +27,8 @@ from src.api.middleware.logging_middleware import LoggingMiddleware
 from src.api.middleware.security import SecurityMiddleware
 from src.api.middleware.compression import CompressionMiddleware
 from src.api.middleware.metrics_middleware import MetricsMiddleware, setup_http_metrics
+from src.api.middleware.ip_whitelist import IPWhitelistMiddleware
+from src.api.middleware.rate_limit import RateLimitMiddleware as RateLimitMiddlewareV2
 from src.infrastructure.observability import (
     CorrelationMiddleware,
     tracing_manager,
@@ -185,6 +187,39 @@ add_compression_middleware(
     exclude_paths={"/health", "/metrics", "/health/metrics", "/api/v1/ws", "/api/v1/observability"}
 )
 
+# Add IP whitelist middleware (only in production)
+if settings.is_production or settings.app_env == "staging":
+    app.add_middleware(
+        IPWhitelistMiddleware,
+        enabled=True,
+        excluded_paths=[
+            "/health",
+            "/healthz",
+            "/ping",
+            "/ready",
+            "/docs",
+            "/redoc",
+            "/openapi.json",
+            "/metrics",
+            "/static",
+            "/favicon.ico",
+            "/_next",
+            "/api/v1/auth/login",
+            "/api/v1/auth/register", 
+            "/api/v1/auth/refresh",
+            "/api/v1/public",
+            "/api/v1/webhooks/incoming"
+        ],
+        strict_mode=False  # Allow requests if IP can't be determined
+    )
+
+# Add rate limiting middleware v2
+app.add_middleware(
+    RateLimitMiddlewareV2,
+    default_tier="free",
+    strategy="sliding_window"
+)
+
 
 # Custom OpenAPI schema
 def custom_openapi():
@@ -361,6 +396,22 @@ app.include_router(
     tags=["Notifications"]
 )
 
+# Import and include admin routes
+from src.api.routes.admin import ip_whitelist as admin_ip_whitelist
+from src.api.routes import api_keys
+
+app.include_router(
+    admin_ip_whitelist.router,
+    prefix="/api/v1/admin",
+    tags=["Admin - IP Whitelist"]
+)
+
+app.include_router(
+    api_keys.router,
+    prefix="/api/v1",
+    tags=["API Keys"]
+)
+
 
 # Global exception handler
 @app.exception_handler(CidadaoAIError)

src/api/middleware/__init__.py

@@ -0,0 +1,24 @@
+"""
+API middleware package.
+
+This module exports all middleware classes for easy import.
+"""
+
+from .rate_limit import RateLimitMiddleware, rate_limit
+from .webhook_verification import (
+    WebhookVerificationMiddleware,
+    WebhookSigner,
+    create_webhook_signature,
+    verify_webhook_signature
+)
+from .ip_whitelist import IPWhitelistMiddleware
+
+__all__ = [
+    "RateLimitMiddleware",
+    "rate_limit",
+    "WebhookVerificationMiddleware", 
+    "WebhookSigner",
+    "create_webhook_signature",
+    "verify_webhook_signature",
+    "IPWhitelistMiddleware"
+]

src/api/middleware/ip_whitelist.py

@@ -0,0 +1,261 @@
+"""
+Module: api.middleware.ip_whitelist
+Description: IP whitelist middleware for production environments
+Author: Anderson H. Silva
+Date: 2025-01-25
+License: Proprietary - All rights reserved
+"""
+
+from typing import Optional, List, Set
+import ipaddress
+
+from fastapi import Request, HTTPException, status
+from starlette.middleware.base import BaseHTTPMiddleware
+from starlette.responses import JSONResponse
+
+from src.core import get_logger
+from src.core.config import settings
+from src.services.ip_whitelist_service import ip_whitelist_service
+from src.db.session import get_session
+
+logger = get_logger(__name__)
+
+
+class IPWhitelistMiddleware(BaseHTTPMiddleware):
+    """
+    IP whitelist middleware for production environments.
+    
+    Features:
+    - Environment-based activation
+    - Path exclusions
+    - Multiple IP extraction methods
+    - Performance optimization with caching
+    """
+    
+    def __init__(
+        self,
+        app,
+        enabled: Optional[bool] = None,
+        excluded_paths: Optional[List[str]] = None,
+        strict_mode: bool = True
+    ):
+        """
+        Initialize IP whitelist middleware.
+        
+        Args:
+            app: FastAPI application
+            enabled: Force enable/disable (None = auto based on environment)
+            excluded_paths: Paths to exclude from whitelist check
+            strict_mode: If True, reject if can't determine IP
+        """
+        super().__init__(app)
+        
+        # Auto-enable in production if not specified
+        if enabled is None:
+            self.enabled = settings.is_production
+        else:
+            self.enabled = enabled
+            
+        self.excluded_paths = set(excluded_paths or self._get_default_excluded_paths())
+        self.strict_mode = strict_mode
+        
+        logger.info(
+            "ip_whitelist_middleware_initialized",
+            enabled=self.enabled,
+            environment=settings.app_env,
+            strict_mode=self.strict_mode
+        )
+        
+    async def dispatch(self, request: Request, call_next):
+        """Process request with IP whitelist check."""
+        # Skip if disabled
+        if not self.enabled:
+            return await call_next(request)
+            
+        # Skip excluded paths
+        if self._should_skip(request.url.path):
+            return await call_next(request)
+        
+        try:
+            # Extract client IP
+            client_ip = self._get_client_ip(request)
+            
+            if not client_ip:
+                if self.strict_mode:
+                    logger.warning(
+                        "ip_whitelist_no_client_ip",
+                        path=request.url.path,
+                        headers=dict(request.headers)
+                    )
+                    return JSONResponse(
+                        status_code=status.HTTP_403_FORBIDDEN,
+                        content={
+                            "detail": "Client IP could not be determined",
+                            "error": "IP_NOT_DETERMINED"
+                        }
+                    )
+                else:
+                    # Allow through if can't determine IP
+                    logger.debug("ip_whitelist_no_client_ip_allowing")
+                    return await call_next(request)
+            
+            # Check whitelist
+            async with get_session() as session:
+                is_whitelisted = await ip_whitelist_service.check_ip(
+                    session,
+                    client_ip,
+                    environment=settings.app_env
+                )
+            
+            if not is_whitelisted:
+                logger.warning(
+                    "ip_whitelist_access_denied",
+                    client_ip=client_ip,
+                    path=request.url.path,
+                    method=request.method
+                )
+                
+                return JSONResponse(
+                    status_code=status.HTTP_403_FORBIDDEN,
+                    content={
+                        "detail": "Access denied: IP not whitelisted",
+                        "error": "IP_NOT_WHITELISTED"
+                    }
+                )
+            
+            # IP is whitelisted, proceed
+            request.state.client_ip = client_ip
+            request.state.ip_whitelisted = True
+            
+            return await call_next(request)
+            
+        except Exception as e:
+            logger.error(
+                "ip_whitelist_error",
+                error=str(e),
+                exc_info=True
+            )
+            
+            # In case of error, fail open or closed based on strict mode
+            if self.strict_mode:
+                return JSONResponse(
+                    status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+                    content={
+                        "detail": "IP whitelist check failed",
+                        "error": "WHITELIST_CHECK_ERROR"
+                    }
+                )
+            else:
+                # Allow through on error
+                return await call_next(request)
+    
+    def _get_default_excluded_paths(self) -> List[str]:
+        """Get default paths to exclude from whitelist."""
+        return [
+            # Health checks
+            "/health",
+            "/healthz",
+            "/ping",
+            "/ready",
+            
+            # Documentation
+            "/docs",
+            "/redoc",
+            "/openapi.json",
+            
+            # Metrics
+            "/metrics",
+            
+            # Static assets
+            "/static",
+            "/favicon.ico",
+            "/_next",
+            
+            # Public endpoints
+            "/api/v1/auth/login",
+            "/api/v1/auth/register",
+            "/api/v1/auth/refresh",
+            "/api/v1/public",
+            
+            # Webhook endpoints (they have their own auth)
+            "/api/v1/webhooks/incoming"
+        ]
+    
+    def _should_skip(self, path: str) -> bool:
+        """Check if path should skip whitelist check."""
+        # Exact match
+        if path in self.excluded_paths:
+            return True
+            
+        # Prefix match
+        for excluded in self.excluded_paths:
+            if excluded.endswith("*") and path.startswith(excluded[:-1]):
+                return True
+            if path.startswith(excluded + "/"):
+                return True
+                
+        return False
+    
+    def _get_client_ip(self, request: Request) -> Optional[str]:
+        """
+        Extract client IP from request.
+        
+        Tries multiple methods in order:
+        1. X-Real-IP header
+        2. X-Forwarded-For header
+        3. CloudFlare headers
+        4. Direct client connection
+        """
+        # Try X-Real-IP first (nginx)
+        real_ip = request.headers.get("X-Real-IP")
+        if real_ip and self._is_valid_ip(real_ip):
+            return real_ip
+            
+        # Try X-Forwarded-For (proxy chains)
+        forwarded_for = request.headers.get("X-Forwarded-For")
+        if forwarded_for:
+            # Take the first IP in the chain
+            ips = [ip.strip() for ip in forwarded_for.split(",")]
+            for ip in ips:
+                if self._is_valid_ip(ip):
+                    return ip
+        
+        # Try CloudFlare headers
+        cf_ip = request.headers.get("CF-Connecting-IP")
+        if cf_ip and self._is_valid_ip(cf_ip):
+            return cf_ip
+            
+        # Try True-Client-IP (Akamai, CloudFlare Enterprise)
+        true_client_ip = request.headers.get("True-Client-IP")
+        if true_client_ip and self._is_valid_ip(true_client_ip):
+            return true_client_ip
+            
+        # Try Fastly header
+        fastly_ip = request.headers.get("Fastly-Client-IP")
+        if fastly_ip and self._is_valid_ip(fastly_ip):
+            return fastly_ip
+            
+        # Fallback to direct connection
+        if request.client and request.client.host:
+            return request.client.host
+            
+        return None
+    
+    def _is_valid_ip(self, ip: str) -> bool:
+        """Check if string is a valid IP address."""
+        if not ip:
+            return False
+            
+        try:
+            # Validate IP
+            ipaddress.ip_address(ip)
+            
+            # Reject private/local IPs unless in development
+            if not settings.is_development:
+                ip_obj = ipaddress.ip_address(ip)
+                if ip_obj.is_private or ip_obj.is_loopback:
+                    return False
+                    
+            return True
+        except ValueError:
+            return False

src/api/middleware/webhook_verification.py

@@ -0,0 +1,295 @@
+"""
+Module: api.middleware.webhook_verification
+Description: Webhook signature verification middleware
+Author: Anderson H. Silva
+Date: 2025-01-25
+License: Proprietary - All rights reserved
+"""
+
+import hmac
+import hashlib
+import time
+from typing import Optional, Dict, Any
+
+from fastapi import Request, HTTPException, status
+from starlette.middleware.base import BaseHTTPMiddleware
+from starlette.responses import JSONResponse
+
+from src.core import get_logger
+
+logger = get_logger(__name__)
+
+
+class WebhookVerificationMiddleware(BaseHTTPMiddleware):
+    """
+    Middleware for verifying incoming webhook signatures.
+    
+    Protects endpoints that receive webhooks from external services.
+    """
+    
+    def __init__(
+        self,
+        app,
+        webhook_paths: Optional[Dict[str, str]] = None,
+        max_timestamp_age: int = 300  # 5 minutes
+    ):
+        """
+        Initialize webhook verification middleware.
+        
+        Args:
+            app: FastAPI application
+            webhook_paths: Dict of path -> secret mapping
+            max_timestamp_age: Maximum age of timestamp in seconds
+        """
+        super().__init__(app)
+        self.webhook_paths = webhook_paths or {}
+        self.max_timestamp_age = max_timestamp_age
+        
+    async def dispatch(self, request: Request, call_next):
+        """Process request with webhook verification."""
+        # Check if this is a webhook path
+        if request.url.path not in self.webhook_paths:
+            return await call_next(request)
+        
+        # Get the secret for this path
+        secret = self.webhook_paths[request.url.path]
+        
+        try:
+            # Read body
+            body = await request.body()
+            
+            # Verify signature
+            if not self._verify_signature(request, body, secret):
+                logger.warning(
+                    "webhook_signature_verification_failed",
+                    path=request.url.path,
+                    headers=dict(request.headers)
+                )
+                
+                return JSONResponse(
+                    status_code=status.HTTP_401_UNAUTHORIZED,
+                    content={
+                        "detail": "Invalid webhook signature",
+                        "error": "INVALID_SIGNATURE"
+                    }
+                )
+            
+            # Verify timestamp if present
+            if not self._verify_timestamp(request):
+                logger.warning(
+                    "webhook_timestamp_verification_failed",
+                    path=request.url.path
+                )
+                
+                return JSONResponse(
+                    status_code=status.HTTP_401_UNAUTHORIZED,
+                    content={
+                        "detail": "Webhook timestamp too old",
+                        "error": "TIMESTAMP_TOO_OLD"
+                    }
+                )
+            
+            # Store raw body for handler
+            request.state.webhook_body = body
+            
+            # Process request
+            return await call_next(request)
+            
+        except Exception as e:
+            logger.error(
+                "webhook_verification_error",
+                error=str(e),
+                exc_info=True
+            )
+            
+            return JSONResponse(
+                status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+                content={
+                    "detail": "Webhook verification error",
+                    "error": "VERIFICATION_ERROR"
+                }
+            )
+    
+    def _verify_signature(
+        self,
+        request: Request,
+        body: bytes,
+        secret: str
+    ) -> bool:
+        """Verify webhook signature."""
+        # Get signature header - support multiple formats
+        signature_header = (
+            request.headers.get("X-Cidadao-Signature") or
+            request.headers.get("X-Webhook-Signature") or
+            request.headers.get("X-Hub-Signature-256")  # GitHub format
+        )
+        
+        if not signature_header:
+            logger.debug("No signature header found")
+            return False
+        
+        # Parse signature
+        if "=" in signature_header:
+            algorithm, signature = signature_header.split("=", 1)
+        else:
+            algorithm = "sha256"
+            signature = signature_header
+        
+        # Calculate expected signature
+        if algorithm == "sha256":
+            expected = hmac.new(
+                secret.encode(),
+                body,
+                hashlib.sha256
+            ).hexdigest()
+        elif algorithm == "sha1":
+            expected = hmac.new(
+                secret.encode(),
+                body,
+                hashlib.sha1
+            ).hexdigest()
+        else:
+            logger.warning(f"Unsupported signature algorithm: {algorithm}")
+            return False
+        
+        # Compare signatures
+        return hmac.compare_digest(signature, expected)
+    
+    def _verify_timestamp(self, request: Request) -> bool:
+        """Verify webhook timestamp is recent."""
+        timestamp_header = (
+            request.headers.get("X-Cidadao-Timestamp") or
+            request.headers.get("X-Webhook-Timestamp")
+        )
+        
+        if not timestamp_header:
+            # No timestamp to verify
+            return True
+        
+        try:
+            # Parse timestamp
+            if timestamp_header.isdigit():
+                # Unix timestamp
+                webhook_time = float(timestamp_header)
+            else:
+                # ISO format
+                from dateutil.parser import parse
+                webhook_time = parse(timestamp_header).timestamp()
+            
+            # Check age
+            current_time = time.time()
+            age = abs(current_time - webhook_time)
+            
+            return age <= self.max_timestamp_age
+            
+        except Exception as e:
+            logger.error(f"Failed to parse timestamp: {e}")
+            return False
+
+
+def create_webhook_signature(
+    payload: bytes,
+    secret: str,
+    algorithm: str = "sha256"
+) -> str:
+    """
+    Create webhook signature for outgoing webhooks.
+    
+    Args:
+        payload: Request body
+        secret: Webhook secret
+        algorithm: Hash algorithm (sha256, sha1)
+        
+    Returns:
+        Signature string with format "algorithm=signature"
+    """
+    if algorithm == "sha256":
+        signature = hmac.new(
+            secret.encode(),
+            payload,
+            hashlib.sha256
+        ).hexdigest()
+    elif algorithm == "sha1":
+        signature = hmac.new(
+            secret.encode(),
+            payload,
+            hashlib.sha1
+        ).hexdigest()
+    else:
+        raise ValueError(f"Unsupported algorithm: {algorithm}")
+    
+    return f"{algorithm}={signature}"
+
+
+def verify_webhook_signature(
+    signature: str,
+    payload: bytes,
+    secret: str
+) -> bool:
+    """
+    Verify webhook signature.
+    
+    Args:
+        signature: Signature header value
+        payload: Request body
+        secret: Webhook secret
+        
+    Returns:
+        True if signature is valid
+    """
+    try:
+        # Parse signature
+        if "=" in signature:
+            algorithm, sig = signature.split("=", 1)
+        else:
+            algorithm = "sha256"
+            sig = signature
+        
+        # Generate expected signature
+        expected = create_webhook_signature(payload, secret, algorithm)
+        
+        # Extract just the signature part
+        if "=" in expected:
+            _, expected_sig = expected.split("=", 1)
+        else:
+            expected_sig = expected
+        
+        # Compare
+        return hmac.compare_digest(sig, expected_sig)
+        
+    except Exception as e:
+        logger.error(f"Signature verification error: {e}")
+        return False
+
+
+class WebhookSigner:
+    """Helper class for signing webhook requests."""
+    
+    def __init__(self, secret: str, algorithm: str = "sha256"):
+        """Initialize webhook signer."""
+        self.secret = secret
+        self.algorithm = algorithm
+        
+    def sign(self, payload: bytes) -> Dict[str, str]:
+        """
+        Generate webhook headers with signature.
+        
+        Args:
+            payload: Request body
+            
+        Returns:
+            Dict of headers to include in request
+        """
+        signature = create_webhook_signature(
+            payload,
+            self.secret,
+            self.algorithm
+        )
+        
+        timestamp = str(int(time.time()))
+        
+        return {
+            "X-Cidadao-Signature": signature,
+            "X-Cidadao-Timestamp": timestamp,
+            "X-Cidadao-Algorithm": self.algorithm
+        }

src/api/routes/admin/__init__.py

@@ -0,0 +1 @@
+"""Admin routes package."""

src/api/routes/admin/ip_whitelist.py

@@ -0,0 +1,377 @@
+"""
+Module: api.routes.admin.ip_whitelist
+Description: Admin routes for managing IP whitelist
+Author: Anderson H. Silva
+Date: 2025-01-25
+License: Proprietary - All rights reserved
+"""
+
+from typing import List, Optional, Dict, Any
+from datetime import datetime
+import ipaddress
+
+from fastapi import APIRouter, Depends, HTTPException, status, Query
+from pydantic import BaseModel, Field, field_validator
+
+from src.core import get_logger
+from src.api.dependencies import require_admin, get_db
+from src.services.ip_whitelist_service import ip_whitelist_service, IPWhitelist
+from src.core.config import settings
+
+logger = get_logger(__name__)
+
+router = APIRouter(prefix="/ip-whitelist", tags=["Admin - IP Whitelist"])
+
+
+class IPWhitelistRequest(BaseModel):
+    """Request to add IP to whitelist."""
+    ip_address: str = Field(..., description="IP address or CIDR range")
+    description: Optional[str] = Field(None, description="Description of the IP")
+    environment: str = Field(default="production", description="Environment")
+    expires_at: Optional[datetime] = Field(None, description="Expiration date")
+    metadata: Optional[Dict[str, Any]] = Field(None, description="Additional metadata")
+    
+    @field_validator("ip_address")
+    @classmethod
+    def validate_ip(cls, v: str) -> str:
+        """Validate IP address or CIDR."""
+        try:
+            # Try as single IP
+            ipaddress.ip_address(v)
+            return v
+        except ValueError:
+            # Try as CIDR
+            try:
+                ipaddress.ip_network(v, strict=False)
+                return v
+            except ValueError:
+                raise ValueError(f"Invalid IP address or CIDR: {v}")
+    
+    @field_validator("environment")
+    @classmethod
+    def validate_environment(cls, v: str) -> str:
+        """Validate environment."""
+        allowed = ["development", "staging", "production", "testing"]
+        if v not in allowed:
+            raise ValueError(f"Environment must be one of {allowed}")
+        return v
+
+
+class IPWhitelistUpdateRequest(BaseModel):
+    """Request to update IP whitelist entry."""
+    active: Optional[bool] = None
+    description: Optional[str] = None
+    expires_at: Optional[datetime] = None
+    metadata: Optional[Dict[str, Any]] = None
+
+
+class IPWhitelistResponse(BaseModel):
+    """IP whitelist entry response."""
+    id: str
+    ip_address: str
+    description: Optional[str]
+    environment: str
+    active: bool
+    is_cidr: bool
+    cidr_prefix: Optional[int]
+    created_by: str
+    created_at: datetime
+    expires_at: Optional[datetime]
+    metadata: Dict[str, Any]
+    is_expired: bool
+    
+    @classmethod
+    def from_model(cls, model: IPWhitelist) -> "IPWhitelistResponse":
+        """Create response from model."""
+        return cls(
+            id=model.id,
+            ip_address=model.ip_address,
+            description=model.description,
+            environment=model.environment,
+            active=model.active,
+            is_cidr=model.is_cidr,
+            cidr_prefix=model.cidr_prefix,
+            created_by=model.created_by,
+            created_at=model.created_at,
+            expires_at=model.expires_at,
+            metadata=model.metadata,
+            is_expired=model.is_expired()
+        )
+
+
+@router.post("/add", response_model=IPWhitelistResponse)
+async def add_ip_to_whitelist(
+    request: IPWhitelistRequest,
+    admin_user=Depends(require_admin),
+    db=Depends(get_db)
+):
+    """
+    Add IP address or CIDR range to whitelist.
+    
+    Requires admin privileges.
+    """
+    try:
+        is_cidr = "/" in request.ip_address
+        
+        entry = await ip_whitelist_service.add_ip(
+            session=db,
+            ip_address=request.ip_address,
+            created_by=admin_user.get("email", "admin"),
+            description=request.description,
+            environment=request.environment,
+            expires_at=request.expires_at,
+            is_cidr=is_cidr,
+            metadata=request.metadata
+        )
+        
+        logger.info(
+            "admin_ip_whitelist_added",
+            admin=admin_user.get("email"),
+            ip=request.ip_address,
+            environment=request.environment
+        )
+        
+        return IPWhitelistResponse.from_model(entry)
+        
+    except ValueError as e:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=str(e)
+        )
+    except Exception as e:
+        logger.error(
+            "admin_ip_whitelist_add_error",
+            error=str(e),
+            exc_info=True
+        )
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="Failed to add IP to whitelist"
+        )
+
+
+@router.delete("/remove/{ip_address}")
+async def remove_ip_from_whitelist(
+    ip_address: str,
+    environment: str = Query(default="production"),
+    admin_user=Depends(require_admin),
+    db=Depends(get_db)
+):
+    """
+    Remove IP from whitelist.
+    
+    Requires admin privileges.
+    """
+    removed = await ip_whitelist_service.remove_ip(
+        session=db,
+        ip_address=ip_address,
+        environment=environment
+    )
+    
+    if not removed:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"IP not found in whitelist: {ip_address}"
+        )
+    
+    logger.info(
+        "admin_ip_whitelist_removed",
+        admin=admin_user.get("email"),
+        ip=ip_address,
+        environment=environment
+    )
+    
+    return {"status": "removed", "ip_address": ip_address}
+
+
+@router.get("/list", response_model=List[IPWhitelistResponse])
+async def list_whitelisted_ips(
+    environment: str = Query(default="production"),
+    include_expired: bool = Query(default=False),
+    admin_user=Depends(require_admin),
+    db=Depends(get_db)
+):
+    """
+    List all whitelisted IPs.
+    
+    Requires admin privileges.
+    """
+    entries = await ip_whitelist_service.list_ips(
+        session=db,
+        environment=environment,
+        include_expired=include_expired
+    )
+    
+    return [IPWhitelistResponse.from_model(entry) for entry in entries]
+
+
+@router.get("/check/{ip_address}")
+async def check_ip_whitelist(
+    ip_address: str,
+    environment: str = Query(default="production"),
+    admin_user=Depends(require_admin),
+    db=Depends(get_db)
+):
+    """
+    Check if IP is whitelisted.
+    
+    Requires admin privileges.
+    """
+    try:
+        # Validate IP
+        ipaddress.ip_address(ip_address)
+    except ValueError:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="Invalid IP address format"
+        )
+    
+    is_whitelisted = await ip_whitelist_service.check_ip(
+        session=db,
+        ip_address=ip_address,
+        environment=environment
+    )
+    
+    return {
+        "ip_address": ip_address,
+        "environment": environment,
+        "is_whitelisted": is_whitelisted
+    }
+
+
+@router.put("/update/{ip_address}", response_model=IPWhitelistResponse)
+async def update_whitelist_entry(
+    ip_address: str,
+    request: IPWhitelistUpdateRequest,
+    environment: str = Query(default="production"),
+    admin_user=Depends(require_admin),
+    db=Depends(get_db)
+):
+    """
+    Update whitelist entry.
+    
+    Requires admin privileges.
+    """
+    entry = await ip_whitelist_service.update_ip(
+        session=db,
+        ip_address=ip_address,
+        environment=environment,
+        active=request.active,
+        description=request.description,
+        expires_at=request.expires_at,
+        metadata=request.metadata
+    )
+    
+    if not entry:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"IP not found in whitelist: {ip_address}"
+        )
+    
+    logger.info(
+        "admin_ip_whitelist_updated",
+        admin=admin_user.get("email"),
+        ip=ip_address,
+        active=entry.active
+    )
+    
+    return IPWhitelistResponse.from_model(entry)
+
+
+@router.post("/cleanup")
+async def cleanup_expired_entries(
+    environment: Optional[str] = None,
+    admin_user=Depends(require_admin),
+    db=Depends(get_db)
+):
+    """
+    Remove expired whitelist entries.
+    
+    Requires admin privileges.
+    """
+    count = await ip_whitelist_service.cleanup_expired(
+        session=db,
+        environment=environment
+    )
+    
+    logger.info(
+        "admin_ip_whitelist_cleanup",
+        admin=admin_user.get("email"),
+        removed=count,
+        environment=environment
+    )
+    
+    return {
+        "status": "cleaned",
+        "removed_count": count,
+        "environment": environment
+    }
+
+
+@router.post("/initialize-defaults")
+async def initialize_default_whitelist(
+    admin_user=Depends(require_admin),
+    db=Depends(get_db)
+):
+    """
+    Initialize default whitelist entries for current environment.
+    
+    Requires admin privileges.
+    """
+    count = await ip_whitelist_service.initialize_defaults(
+        session=db,
+        created_by=admin_user.get("email", "admin")
+    )
+    
+    logger.info(
+        "admin_ip_whitelist_defaults_initialized",
+        admin=admin_user.get("email"),
+        count=count,
+        environment=settings.app_env
+    )
+    
+    return {
+        "status": "initialized",
+        "added_count": count,
+        "environment": settings.app_env,
+        "defaults": ip_whitelist_service.get_default_whitelist()
+    }
+
+
+@router.get("/stats")
+async def get_whitelist_stats(
+    admin_user=Depends(require_admin),
+    db=Depends(get_db)
+):
+    """
+    Get whitelist statistics.
+    
+    Requires admin privileges.
+    """
+    environments = ["development", "staging", "production"]
+    stats = {}
+    
+    for env in environments:
+        entries = await ip_whitelist_service.list_ips(
+            session=db,
+            environment=env,
+            include_expired=True
+        )
+        
+        active = sum(1 for e in entries if e.active and not e.is_expired())
+        expired = sum(1 for e in entries if e.is_expired())
+        cidr_ranges = sum(1 for e in entries if e.is_cidr)
+        
+        stats[env] = {
+            "total": len(entries),
+            "active": active,
+            "expired": expired,
+            "inactive": sum(1 for e in entries if not e.active),
+            "cidr_ranges": cidr_ranges,
+            "single_ips": len(entries) - cidr_ranges
+        }
+    
+    return {
+        "statistics": stats,
+        "current_environment": settings.app_env
+    }

src/api/routes/webhooks.py

@@ -0,0 +1,407 @@
+"""
+Module: api.routes.webhooks
+Description: Webhook endpoints for receiving external events
+Author: Anderson H. Silva
+Date: 2025-01-25
+License: Proprietary - All rights reserved
+"""
+
+from typing import Dict, Any, Optional
+from datetime import datetime
+
+from fastapi import APIRouter, Request, Depends, HTTPException, status, BackgroundTasks
+from pydantic import BaseModel, Field
+
+from src.core import get_logger
+from src.api.dependencies import get_current_user
+from src.services.webhook_service import WebhookConfig, WebhookEvent, webhook_service
+from src.api.middleware.webhook_verification import verify_webhook_signature
+
+logger = get_logger(__name__)
+
+router = APIRouter(prefix="/webhooks", tags=["Webhooks"])
+
+
+class IncomingWebhookPayload(BaseModel):
+    """Generic incoming webhook payload."""
+    event: str
+    timestamp: Optional[datetime] = None
+    data: Dict[str, Any]
+    signature: Optional[str] = None
+
+
+class WebhookRegistrationRequest(BaseModel):
+    """Request to register a new webhook."""
+    url: str = Field(..., description="Webhook endpoint URL")
+    events: Optional[list[str]] = Field(None, description="Events to subscribe to (None = all)")
+    secret: Optional[str] = Field(None, description="Webhook secret for HMAC signing")
+    headers: Optional[Dict[str, str]] = Field(None, description="Custom headers")
+    active: bool = Field(True, description="Whether webhook is active")
+
+
+class WebhookTestRequest(BaseModel):
+    """Request to test a webhook."""
+    url: str = Field(..., description="Webhook URL to test")
+    secret: Optional[str] = Field(None, description="Webhook secret if any")
+
+
+@router.post("/incoming/github")
+async def receive_github_webhook(
+    request: Request,
+    background_tasks: BackgroundTasks
+):
+    """
+    Receive webhooks from GitHub.
+    
+    Requires webhook signature verification.
+    """
+    # Get raw body from request state (set by verification middleware)
+    body = getattr(request.state, "webhook_body", None)
+    if not body:
+        body = await request.body()
+    
+    # Parse event type
+    event_type = request.headers.get("X-GitHub-Event", "unknown")
+    
+    # Parse payload
+    try:
+        import json
+        payload = json.loads(body)
+    except Exception as e:
+        logger.error("Failed to parse GitHub webhook", error=str(e))
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="Invalid payload format"
+        )
+    
+    # Log webhook received
+    logger.info(
+        "github_webhook_received",
+        event=event_type,
+        repository=payload.get("repository", {}).get("full_name"),
+        action=payload.get("action")
+    )
+    
+    # Process webhook asynchronously
+    background_tasks.add_task(
+        process_github_webhook,
+        event_type,
+        payload
+    )
+    
+    return {"status": "accepted", "event": event_type}
+
+
+@router.post("/incoming/generic/{webhook_id}")
+async def receive_generic_webhook(
+    webhook_id: str,
+    request: Request,
+    payload: IncomingWebhookPayload,
+    background_tasks: BackgroundTasks
+):
+    """
+    Receive generic webhooks with configurable verification.
+    
+    The webhook_id should match a configured incoming webhook.
+    """
+    # Verify webhook ID exists and get configuration
+    # In production, this would look up from database
+    webhook_config = get_incoming_webhook_config(webhook_id)
+    
+    if not webhook_config:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"Webhook configuration not found: {webhook_id}"
+        )
+    
+    # Verify signature if secret is configured
+    if webhook_config.get("secret"):
+        body = await request.body()
+        signature = request.headers.get("X-Webhook-Signature")
+        
+        if not signature:
+            raise HTTPException(
+                status_code=status.HTTP_401_UNAUTHORIZED,
+                detail="Missing webhook signature"
+            )
+        
+        if not verify_webhook_signature(signature, body, webhook_config["secret"]):
+            raise HTTPException(
+                status_code=status.HTTP_401_UNAUTHORIZED,
+                detail="Invalid webhook signature"
+            )
+    
+    # Log webhook
+    logger.info(
+        "generic_webhook_received",
+        webhook_id=webhook_id,
+        event=payload.event,
+        timestamp=payload.timestamp
+    )
+    
+    # Process asynchronously
+    background_tasks.add_task(
+        process_generic_webhook,
+        webhook_id,
+        payload
+    )
+    
+    return {
+        "status": "accepted",
+        "webhook_id": webhook_id,
+        "event": payload.event
+    }
+
+
+@router.post("/register")
+async def register_webhook(
+    request: WebhookRegistrationRequest,
+    current_user=Depends(get_current_user)
+):
+    """
+    Register a new outgoing webhook.
+    
+    Requires authentication.
+    """
+    # Convert string events to enum
+    events = None
+    if request.events:
+        try:
+            events = [WebhookEvent(e) for e in request.events]
+        except ValueError as e:
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail=f"Invalid event type: {e}"
+            )
+    
+    # Create webhook config
+    config = WebhookConfig(
+        url=request.url,
+        events=events,
+        secret=request.secret,
+        headers=request.headers,
+        active=request.active
+    )
+    
+    # Register webhook
+    webhook_service.add_webhook(config)
+    
+    logger.info(
+        "webhook_registered",
+        user=current_user.get("email"),
+        url=request.url,
+        events=request.events
+    )
+    
+    return {
+        "status": "registered",
+        "url": request.url,
+        "events": request.events,
+        "active": request.active
+    }
+
+
+@router.delete("/unregister")
+async def unregister_webhook(
+    url: str,
+    current_user=Depends(get_current_user)
+):
+    """
+    Unregister a webhook.
+    
+    Requires authentication.
+    """
+    removed = webhook_service.remove_webhook(url)
+    
+    if not removed:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"Webhook not found: {url}"
+        )
+    
+    logger.info(
+        "webhook_unregistered",
+        user=current_user.get("email"),
+        url=url
+    )
+    
+    return {"status": "unregistered", "url": url}
+
+
+@router.get("/list")
+async def list_webhooks(
+    current_user=Depends(get_current_user)
+):
+    """
+    List all registered outgoing webhooks.
+    
+    Requires authentication.
+    """
+    webhooks = webhook_service.list_webhooks()
+    
+    return {
+        "webhooks": [
+            {
+                "url": str(w.url),
+                "events": [e.value for e in w.events] if w.events else None,
+                "active": w.active,
+                "has_secret": bool(w.secret)
+            }
+            for w in webhooks
+        ],
+        "total": len(webhooks)
+    }
+
+
+@router.post("/test")
+async def test_webhook(
+    request: WebhookTestRequest,
+    background_tasks: BackgroundTasks,
+    current_user=Depends(get_current_user)
+):
+    """
+    Test a webhook endpoint.
+    
+    Sends a test payload to verify webhook is working.
+    """
+    # Create temporary webhook config
+    config = WebhookConfig(
+        url=request.url,
+        secret=request.secret,
+        max_retries=1,
+        timeout=10
+    )
+    
+    # Test webhook
+    delivery = await webhook_service.test_webhook(config)
+    
+    logger.info(
+        "webhook_tested",
+        user=current_user.get("email"),
+        url=request.url,
+        success=delivery.success,
+        status_code=delivery.status_code
+    )
+    
+    return {
+        "url": request.url,
+        "success": delivery.success,
+        "status_code": delivery.status_code,
+        "response": delivery.response_body,
+        "error": delivery.error,
+        "duration_ms": delivery.duration_ms
+    }
+
+
+@router.get("/history")
+async def get_webhook_history(
+    event: Optional[str] = None,
+    url: Optional[str] = None,
+    success: Optional[bool] = None,
+    limit: int = 100,
+    current_user=Depends(get_current_user)
+):
+    """
+    Get webhook delivery history.
+    
+    Requires authentication.
+    """
+    # Convert event string to enum if provided
+    event_enum = None
+    if event:
+        try:
+            event_enum = WebhookEvent(event)
+        except ValueError:
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail=f"Invalid event type: {event}"
+            )
+    
+    history = webhook_service.get_delivery_history(
+        event=event_enum,
+        url=url,
+        success=success,
+        limit=limit
+    )
+    
+    return {
+        "deliveries": [
+            {
+                "webhook_url": d.webhook_url,
+                "event": d.event.value,
+                "timestamp": d.timestamp.isoformat(),
+                "success": d.success,
+                "status_code": d.status_code,
+                "error": d.error,
+                "attempts": d.attempts,
+                "duration_ms": d.duration_ms
+            }
+            for d in history
+        ],
+        "total": len(history)
+    }
+
+
+# Helper functions
+
+def get_incoming_webhook_config(webhook_id: str) -> Optional[Dict[str, Any]]:
+    """Get configuration for incoming webhook."""
+    # In production, this would be from database
+    # For now, return mock config
+    configs = {
+        "test": {
+            "secret": "test-secret",
+            "active": True
+        },
+        "monitoring": {
+            "secret": "monitoring-secret",
+            "active": True
+        }
+    }
+    
+    return configs.get(webhook_id)
+
+
+async def process_github_webhook(event_type: str, payload: Dict[str, Any]):
+    """Process GitHub webhook asynchronously."""
+    try:
+        # Handle different GitHub events
+        if event_type == "push":
+            # Handle code push
+            logger.info("Processing GitHub push event")
+        elif event_type == "pull_request":
+            # Handle pull request
+            logger.info("Processing GitHub pull request event")
+        elif event_type == "issues":
+            # Handle issues
+            logger.info("Processing GitHub issues event")
+        # Add more event handlers as needed
+        
+    except Exception as e:
+        logger.error(
+            "Failed to process GitHub webhook",
+            event=event_type,
+            error=str(e),
+            exc_info=True
+        )
+
+
+async def process_generic_webhook(webhook_id: str, payload: IncomingWebhookPayload):
+    """Process generic webhook asynchronously."""
+    try:
+        # Route to appropriate handler based on webhook_id
+        logger.info(
+            "Processing generic webhook",
+            webhook_id=webhook_id,
+            event=payload.event
+        )
+        
+        # Add specific processing logic here
+        
+    except Exception as e:
+        logger.error(
+            "Failed to process generic webhook",
+            webhook_id=webhook_id,
+            error=str(e),
+            exc_info=True
+        )

src/core/config.py

@@ -178,6 +178,11 @@ class Settings(BaseSettings):
     rate_limit_per_hour: int = Field(default=1000, description="Rate limit per hour")
     rate_limit_per_day: int = Field(default=10000, description="Rate limit per day")
     
+    # IP Whitelist
+    ip_whitelist_enabled: bool = Field(default=True, description="Enable IP whitelist in production")
+    ip_whitelist_strict: bool = Field(default=False, description="Strict mode - reject if IP unknown")
+    ip_whitelist_cache_ttl: int = Field(default=300, description="IP whitelist cache TTL seconds")
+    
     # Celery
     celery_broker_url: str = Field(
         default="redis://localhost:6379/1",

src/services/ip_whitelist_service.py

@@ -0,0 +1,435 @@
+"""
+Module: services.ip_whitelist_service
+Description: IP whitelist management for production environments
+Author: Anderson H. Silva
+Date: 2025-01-25
+License: Proprietary - All rights reserved
+"""
+
+import ipaddress
+from typing import List, Optional, Set, Dict, Any
+from datetime import datetime, timezone
+import json
+
+from src.core import get_logger
+from src.infrastructure.cache import cache_service
+from src.core.config import settings
+from src.models.base import BaseModel
+from sqlalchemy import Column, String, Boolean, DateTime, Integer, JSON
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy import select, delete
+
+logger = get_logger(__name__)
+
+
+class IPWhitelist(BaseModel):
+    """IP whitelist entry model."""
+    __tablename__ = "ip_whitelists"
+    
+    id = Column(String(64), primary_key=True)
+    ip_address = Column(String(45), nullable=False, unique=True)  # IPv4 or IPv6
+    description = Column(String(255))
+    environment = Column(String(20), nullable=False, default="production")
+    active = Column(Boolean, default=True)
+    created_by = Column(String(255), nullable=False)
+    created_at = Column(DateTime(timezone=True), nullable=False, default=lambda: datetime.now(timezone.utc))
+    expires_at = Column(DateTime(timezone=True), nullable=True)
+    metadata = Column(JSON, default=dict)
+    
+    # CIDR support
+    is_cidr = Column(Boolean, default=False)
+    cidr_prefix = Column(Integer, nullable=True)
+    
+    def is_expired(self) -> bool:
+        """Check if whitelist entry is expired."""
+        if not self.expires_at:
+            return False
+        return datetime.now(timezone.utc) > self.expires_at
+    
+    def matches(self, ip: str) -> bool:
+        """Check if IP matches this whitelist entry."""
+        if not self.active or self.is_expired():
+            return False
+            
+        try:
+            if self.is_cidr:
+                # CIDR range check
+                network = ipaddress.ip_network(f"{self.ip_address}/{self.cidr_prefix}")
+                return ipaddress.ip_address(ip) in network
+            else:
+                # Exact match
+                return self.ip_address == ip
+        except ValueError:
+            logger.error(f"Invalid IP address format: {ip}")
+            return False
+
+
+class IPWhitelistService:
+    """Service for managing IP whitelists."""
+    
+    def __init__(self):
+        """Initialize IP whitelist service."""
+        self._cache_key_prefix = "ip_whitelist"
+        self._cache_ttl = 300  # 5 minutes
+        self._whitelist_cache: Optional[Set[str]] = None
+        self._cidr_cache: Optional[List[tuple]] = None
+        self._last_cache_update: Optional[datetime] = None
+        
+    async def add_ip(
+        self,
+        session: AsyncSession,
+        ip_address: str,
+        created_by: str,
+        description: Optional[str] = None,
+        environment: str = "production",
+        expires_at: Optional[datetime] = None,
+        is_cidr: bool = False,
+        metadata: Optional[Dict[str, Any]] = None
+    ) -> IPWhitelist:
+        """Add IP address or CIDR range to whitelist."""
+        try:
+            # Parse and validate IP/CIDR
+            if is_cidr or "/" in ip_address:
+                network = ipaddress.ip_network(ip_address, strict=False)
+                ip_str = str(network.network_address)
+                cidr_prefix = network.prefixlen
+                is_cidr = True
+            else:
+                # Validate single IP
+                ip_obj = ipaddress.ip_address(ip_address)
+                ip_str = str(ip_obj)
+                cidr_prefix = None
+                is_cidr = False
+                
+        except ValueError as e:
+            logger.error(f"Invalid IP address format: {ip_address}")
+            raise ValueError(f"Invalid IP address format: {ip_address}") from e
+        
+        # Check if already exists
+        existing = await session.execute(
+            select(IPWhitelist).where(
+                IPWhitelist.ip_address == ip_str,
+                IPWhitelist.environment == environment
+            )
+        )
+        if existing.scalar_one_or_none():
+            raise ValueError(f"IP address already whitelisted: {ip_str}")
+        
+        # Create whitelist entry
+        entry = IPWhitelist(
+            id=f"{environment}:{ip_str}",
+            ip_address=ip_str,
+            description=description,
+            environment=environment,
+            created_by=created_by,
+            expires_at=expires_at,
+            is_cidr=is_cidr,
+            cidr_prefix=cidr_prefix,
+            metadata=metadata or {}
+        )
+        
+        session.add(entry)
+        await session.commit()
+        
+        # Invalidate cache
+        await self._invalidate_cache()
+        
+        logger.info(
+            "ip_whitelist_added",
+            ip=ip_str,
+            environment=environment,
+            is_cidr=is_cidr,
+            created_by=created_by
+        )
+        
+        return entry
+    
+    async def remove_ip(
+        self,
+        session: AsyncSession,
+        ip_address: str,
+        environment: str = "production"
+    ) -> bool:
+        """Remove IP from whitelist."""
+        result = await session.execute(
+            delete(IPWhitelist).where(
+                IPWhitelist.ip_address == ip_address,
+                IPWhitelist.environment == environment
+            )
+        )
+        await session.commit()
+        
+        if result.rowcount > 0:
+            await self._invalidate_cache()
+            logger.info(
+                "ip_whitelist_removed",
+                ip=ip_address,
+                environment=environment
+            )
+            return True
+        
+        return False
+    
+    async def check_ip(
+        self,
+        session: AsyncSession,
+        ip_address: str,
+        environment: str = "production"
+    ) -> bool:
+        """Check if IP is whitelisted."""
+        # Check cache first
+        cache_key = f"{self._cache_key_prefix}:{environment}:check:{ip_address}"
+        cached = await cache_service.get(cache_key)
+        if cached is not None:
+            return cached
+        
+        # Load whitelist if needed
+        await self._ensure_cache_loaded(session, environment)
+        
+        # Check exact matches first
+        if self._whitelist_cache and ip_address in self._whitelist_cache:
+            await cache_service.set(cache_key, True, ttl=self._cache_ttl)
+            return True
+        
+        # Check CIDR ranges
+        if self._cidr_cache:
+            for cidr_ip, prefix, expires_at in self._cidr_cache:
+                if expires_at and datetime.now(timezone.utc) > expires_at:
+                    continue
+                    
+                try:
+                    network = ipaddress.ip_network(f"{cidr_ip}/{prefix}")
+                    if ipaddress.ip_address(ip_address) in network:
+                        await cache_service.set(cache_key, True, ttl=self._cache_ttl)
+                        return True
+                except ValueError:
+                    continue
+        
+        # Not whitelisted
+        await cache_service.set(cache_key, False, ttl=self._cache_ttl)
+        return False
+    
+    async def list_ips(
+        self,
+        session: AsyncSession,
+        environment: str = "production",
+        include_expired: bool = False
+    ) -> List[IPWhitelist]:
+        """List all whitelisted IPs."""
+        query = select(IPWhitelist).where(
+            IPWhitelist.environment == environment
+        )
+        
+        if not include_expired:
+            now = datetime.now(timezone.utc)
+            query = query.where(
+                (IPWhitelist.expires_at.is_(None)) | 
+                (IPWhitelist.expires_at > now)
+            )
+        
+        result = await session.execute(query)
+        return list(result.scalars().all())
+    
+    async def update_ip(
+        self,
+        session: AsyncSession,
+        ip_address: str,
+        environment: str = "production",
+        active: Optional[bool] = None,
+        description: Optional[str] = None,
+        expires_at: Optional[datetime] = None,
+        metadata: Optional[Dict[str, Any]] = None
+    ) -> Optional[IPWhitelist]:
+        """Update whitelist entry."""
+        result = await session.execute(
+            select(IPWhitelist).where(
+                IPWhitelist.ip_address == ip_address,
+                IPWhitelist.environment == environment
+            )
+        )
+        entry = result.scalar_one_or_none()
+        
+        if not entry:
+            return None
+        
+        if active is not None:
+            entry.active = active
+        if description is not None:
+            entry.description = description
+        if expires_at is not None:
+            entry.expires_at = expires_at
+        if metadata is not None:
+            entry.metadata = metadata
+        
+        await session.commit()
+        await self._invalidate_cache()
+        
+        logger.info(
+            "ip_whitelist_updated",
+            ip=ip_address,
+            environment=environment,
+            active=entry.active
+        )
+        
+        return entry
+    
+    async def cleanup_expired(
+        self,
+        session: AsyncSession,
+        environment: Optional[str] = None
+    ) -> int:
+        """Remove expired whitelist entries."""
+        query = delete(IPWhitelist).where(
+            IPWhitelist.expires_at < datetime.now(timezone.utc)
+        )
+        
+        if environment:
+            query = query.where(IPWhitelist.environment == environment)
+        
+        result = await session.execute(query)
+        await session.commit()
+        
+        if result.rowcount > 0:
+            await self._invalidate_cache()
+            logger.info(
+                "ip_whitelist_cleanup",
+                removed=result.rowcount,
+                environment=environment
+            )
+        
+        return result.rowcount
+    
+    async def _ensure_cache_loaded(
+        self,
+        session: AsyncSession,
+        environment: str
+    ) -> None:
+        """Ensure whitelist is loaded in cache."""
+        # Check if cache is still valid
+        if (
+            self._last_cache_update and
+            (datetime.now(timezone.utc) - self._last_cache_update).total_seconds() < self._cache_ttl
+        ):
+            return
+        
+        # Load from database
+        now = datetime.now(timezone.utc)
+        result = await session.execute(
+            select(IPWhitelist).where(
+                IPWhitelist.environment == environment,
+                IPWhitelist.active == True,
+                (IPWhitelist.expires_at.is_(None)) | (IPWhitelist.expires_at > now)
+            )
+        )
+        
+        entries = result.scalars().all()
+        
+        # Separate exact IPs and CIDR ranges
+        self._whitelist_cache = set()
+        self._cidr_cache = []
+        
+        for entry in entries:
+            if entry.is_cidr:
+                self._cidr_cache.append((
+                    entry.ip_address,
+                    entry.cidr_prefix,
+                    entry.expires_at
+                ))
+            else:
+                self._whitelist_cache.add(entry.ip_address)
+        
+        self._last_cache_update = datetime.now(timezone.utc)
+        
+        logger.debug(
+            "ip_whitelist_cache_loaded",
+            environment=environment,
+            exact_ips=len(self._whitelist_cache),
+            cidr_ranges=len(self._cidr_cache)
+        )
+    
+    async def _invalidate_cache(self) -> None:
+        """Invalidate the whitelist cache."""
+        self._whitelist_cache = None
+        self._cidr_cache = None
+        self._last_cache_update = None
+        
+        # Clear Redis cache patterns
+        pattern = f"{self._cache_key_prefix}:*"
+        await cache_service.delete_pattern(pattern)
+    
+    def get_default_whitelist(self) -> List[str]:
+        """Get default whitelist based on environment."""
+        defaults = []
+        
+        # Always allow localhost
+        defaults.extend([
+            "127.0.0.1",
+            "::1",
+            "localhost"
+        ])
+        
+        # Development environment
+        if settings.is_development:
+            defaults.extend([
+                "10.0.0.0/8",    # Private network
+                "172.16.0.0/12", # Private network
+                "192.168.0.0/16" # Private network
+            ])
+        
+        # Production environment - add known services
+        if settings.is_production:
+            # Vercel IPs (example - would need real ranges)
+            defaults.extend([
+                "76.76.21.0/24",   # Vercel edge network (example)
+                "76.223.0.0/16"    # Vercel edge network (example)
+            ])
+            
+            # HuggingFace Spaces IPs (example - would need real ranges)
+            defaults.extend([
+                "34.0.0.0/8",     # Google Cloud (where HF runs)
+                "35.0.0.0/8"      # Google Cloud
+            ])
+            
+            # Monitoring services
+            defaults.extend([
+                "52.0.0.0/8"      # AWS (for monitoring)
+            ])
+        
+        return defaults
+    
+    async def initialize_defaults(
+        self,
+        session: AsyncSession,
+        created_by: str = "system"
+    ) -> int:
+        """Initialize default whitelist entries."""
+        defaults = self.get_default_whitelist()
+        count = 0
+        
+        for ip in defaults:
+            try:
+                is_cidr = "/" in ip
+                await self.add_ip(
+                    session=session,
+                    ip_address=ip,
+                    created_by=created_by,
+                    description="Default whitelist entry",
+                    environment=settings.app_env,
+                    is_cidr=is_cidr
+                )
+                count += 1
+            except ValueError:
+                # Already exists or invalid
+                continue
+        
+        logger.info(
+            "ip_whitelist_defaults_initialized",
+            count=count,
+            environment=settings.app_env
+        )
+        
+        return count
+
+
+# Global instance
+ip_whitelist_service = IPWhitelistService()

tests/unit/services/test_ip_whitelist_service.py

@@ -0,0 +1,300 @@
+"""Tests for IP whitelist service."""
+
+import pytest
+from datetime import datetime, timedelta, timezone
+from sqlalchemy.ext.asyncio import AsyncSession
+from unittest.mock import Mock, AsyncMock, patch
+
+from src.services.ip_whitelist_service import IPWhitelistService, IPWhitelist
+from src.core.config import settings
+
+
+@pytest.fixture
+def ip_whitelist_service():
+    """Create IP whitelist service instance."""
+    return IPWhitelistService()
+
+
+@pytest.fixture
+async def mock_db_session():
+    """Create mock database session."""
+    session = AsyncMock(spec=AsyncSession)
+    session.commit = AsyncMock()
+    session.execute = AsyncMock()
+    session.add = Mock()
+    return session
+
+
+class TestIPWhitelistService:
+    """Test IP whitelist service."""
+    
+    async def test_add_single_ip(self, ip_whitelist_service, mock_db_session):
+        """Test adding a single IP address."""
+        # Mock query result
+        mock_db_session.execute.return_value.scalar_one_or_none.return_value = None
+        
+        # Add IP
+        entry = await ip_whitelist_service.add_ip(
+            session=mock_db_session,
+            ip_address="192.168.1.100",
+            created_by="[email protected]",
+            description="Test IP",
+            environment="production"
+        )
+        
+        # Verify
+        assert entry.ip_address == "192.168.1.100"
+        assert entry.created_by == "[email protected]"
+        assert entry.description == "Test IP"
+        assert entry.environment == "production"
+        assert entry.is_cidr is False
+        assert entry.active is True
+        
+        # Verify database operations
+        mock_db_session.add.assert_called_once()
+        mock_db_session.commit.assert_called_once()
+    
+    async def test_add_cidr_range(self, ip_whitelist_service, mock_db_session):
+        """Test adding a CIDR range."""
+        # Mock query result
+        mock_db_session.execute.return_value.scalar_one_or_none.return_value = None
+        
+        # Add CIDR
+        entry = await ip_whitelist_service.add_ip(
+            session=mock_db_session,
+            ip_address="10.0.0.0/24",
+            created_by="[email protected]",
+            description="Test subnet",
+            environment="production",
+            is_cidr=True
+        )
+        
+        # Verify
+        assert entry.ip_address == "10.0.0.0"
+        assert entry.is_cidr is True
+        assert entry.cidr_prefix == 24
+        assert entry.active is True
+    
+    async def test_add_duplicate_ip_fails(self, ip_whitelist_service, mock_db_session):
+        """Test adding duplicate IP fails."""
+        # Mock existing entry
+        existing = Mock(spec=IPWhitelist)
+        mock_db_session.execute.return_value.scalar_one_or_none.return_value = existing
+        
+        # Try to add duplicate
+        with pytest.raises(ValueError, match="already whitelisted"):
+            await ip_whitelist_service.add_ip(
+                session=mock_db_session,
+                ip_address="192.168.1.100",
+                created_by="[email protected]"
+            )
+    
+    async def test_add_invalid_ip_fails(self, ip_whitelist_service, mock_db_session):
+        """Test adding invalid IP fails."""
+        with pytest.raises(ValueError, match="Invalid IP address format"):
+            await ip_whitelist_service.add_ip(
+                session=mock_db_session,
+                ip_address="not.an.ip.address",
+                created_by="[email protected]"
+            )
+    
+    async def test_check_ip_exact_match(self, ip_whitelist_service, mock_db_session):
+        """Test checking IP with exact match."""
+        # Mock whitelist entries
+        entries = [
+            Mock(
+                ip_address="192.168.1.100",
+                is_cidr=False,
+                active=True,
+                expires_at=None
+            ),
+            Mock(
+                ip_address="10.0.0.0",
+                is_cidr=True,
+                cidr_prefix=24,
+                active=True,
+                expires_at=None
+            )
+        ]
+        
+        mock_db_session.execute.return_value.scalars.return_value.all.return_value = entries
+        
+        # Force cache reload
+        ip_whitelist_service._last_cache_update = None
+        
+        # Check whitelisted IP
+        result = await ip_whitelist_service.check_ip(
+            session=mock_db_session,
+            ip_address="192.168.1.100",
+            environment="production"
+        )
+        assert result is True
+    
+    async def test_check_ip_cidr_match(self, ip_whitelist_service, mock_db_session):
+        """Test checking IP within CIDR range."""
+        # Mock whitelist entries
+        entries = [
+            Mock(
+                ip_address="10.0.0.0",
+                is_cidr=True,
+                cidr_prefix=24,
+                active=True,
+                expires_at=None
+            )
+        ]
+        
+        mock_db_session.execute.return_value.scalars.return_value.all.return_value = entries
+        
+        # Force cache reload
+        ip_whitelist_service._last_cache_update = None
+        
+        # Check IP in range
+        result = await ip_whitelist_service.check_ip(
+            session=mock_db_session,
+            ip_address="10.0.0.50",
+            environment="production"
+        )
+        assert result is True
+        
+        # Check IP outside range
+        result = await ip_whitelist_service.check_ip(
+            session=mock_db_session,
+            ip_address="10.0.1.50",
+            environment="production"
+        )
+        assert result is False
+    
+    async def test_check_ip_expired_entry(self, ip_whitelist_service, mock_db_session):
+        """Test expired entries are ignored."""
+        # Mock expired entry
+        entries = [
+            Mock(
+                ip_address="192.168.1.100",
+                is_cidr=False,
+                active=True,
+                expires_at=datetime.now(timezone.utc) - timedelta(hours=1)
+            )
+        ]
+        
+        mock_db_session.execute.return_value.scalars.return_value.all.return_value = entries
+        
+        # Force cache reload
+        ip_whitelist_service._last_cache_update = None
+        
+        # Check expired IP
+        result = await ip_whitelist_service.check_ip(
+            session=mock_db_session,
+            ip_address="192.168.1.100",
+            environment="production"
+        )
+        assert result is False
+    
+    async def test_remove_ip(self, ip_whitelist_service, mock_db_session):
+        """Test removing IP from whitelist."""
+        # Mock delete result
+        mock_result = Mock()
+        mock_result.rowcount = 1
+        mock_db_session.execute.return_value = mock_result
+        
+        # Remove IP
+        result = await ip_whitelist_service.remove_ip(
+            session=mock_db_session,
+            ip_address="192.168.1.100",
+            environment="production"
+        )
+        
+        assert result is True
+        mock_db_session.commit.assert_called_once()
+    
+    async def test_update_ip(self, ip_whitelist_service, mock_db_session):
+        """Test updating whitelist entry."""
+        # Mock existing entry
+        entry = Mock(spec=IPWhitelist)
+        entry.ip_address = "192.168.1.100"
+        entry.active = True
+        entry.description = "Old description"
+        
+        mock_db_session.execute.return_value.scalar_one_or_none.return_value = entry
+        
+        # Update entry
+        result = await ip_whitelist_service.update_ip(
+            session=mock_db_session,
+            ip_address="192.168.1.100",
+            environment="production",
+            active=False,
+            description="New description"
+        )
+        
+        assert result is not None
+        assert entry.active is False
+        assert entry.description == "New description"
+        mock_db_session.commit.assert_called_once()
+    
+    def test_get_default_whitelist_development(self, ip_whitelist_service):
+        """Test default whitelist for development."""
+        with patch.object(settings, 'is_development', True):
+            defaults = ip_whitelist_service.get_default_whitelist()
+            
+            # Should include localhost and private networks
+            assert "127.0.0.1" in defaults
+            assert "::1" in defaults
+            assert "10.0.0.0/8" in defaults
+            assert "192.168.0.0/16" in defaults
+    
+    def test_get_default_whitelist_production(self, ip_whitelist_service):
+        """Test default whitelist for production."""
+        with patch.object(settings, 'is_production', True):
+            defaults = ip_whitelist_service.get_default_whitelist()
+            
+            # Should include localhost and service IPs
+            assert "127.0.0.1" in defaults
+            assert "::1" in defaults
+            # Should have cloud provider ranges
+            assert any("76." in ip for ip in defaults)  # Vercel
+            assert any("34." in ip for ip in defaults)  # Google Cloud
+    
+    async def test_cleanup_expired(self, ip_whitelist_service, mock_db_session):
+        """Test cleaning up expired entries."""
+        # Mock delete result
+        mock_result = Mock()
+        mock_result.rowcount = 5
+        mock_db_session.execute.return_value = mock_result
+        
+        # Cleanup
+        count = await ip_whitelist_service.cleanup_expired(
+            session=mock_db_session,
+            environment="production"
+        )
+        
+        assert count == 5
+        mock_db_session.commit.assert_called_once()
+    
+    def test_ip_whitelist_model_matches(self):
+        """Test IPWhitelist model matching logic."""
+        # Test exact match
+        entry = IPWhitelist(
+            id="test",
+            ip_address="192.168.1.100",
+            is_cidr=False,
+            active=True,
+            created_by="test"
+        )
+        assert entry.matches("192.168.1.100") is True
+        assert entry.matches("192.168.1.101") is False
+        
+        # Test CIDR match
+        entry_cidr = IPWhitelist(
+            id="test",
+            ip_address="10.0.0.0",
+            is_cidr=True,
+            cidr_prefix=24,
+            active=True,
+            created_by="test"
+        )
+        assert entry_cidr.matches("10.0.0.1") is True
+        assert entry_cidr.matches("10.0.0.255") is True
+        assert entry_cidr.matches("10.0.1.1") is False
+        
+        # Test inactive entry
+        entry.active = False
+        assert entry.matches("192.168.1.100") is False