feat(centrifugo): add channel name validation utilities

chore: bump version to 1.5.117

Author: markolofsen

packages/django_cfg/__init__.py

@@ -32,7 +32,7 @@ class MyConfig(DjangoConfig):
 default_app_config = "django_cfg.apps.DjangoCfgConfig"
 
 # Version information
-__version__ = "1.5.115"
+__version__ = "1.5.117"
 __license__ = "MIT"
 
 # Setup warnings debug early (checks env var only at this point)

packages/django_cfg/apps/integrations/centrifugo/services/channel_validator.py

@@ -0,0 +1,153 @@
+"""
+Centrifugo channel name validation utilities.
+
+Helps detect common mistakes with channel naming that can lead to permission errors.
+"""
+
+import logging
+import re
+import warnings
+from dataclasses import dataclass
+from typing import Optional
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class ChannelValidationResult:
+    """Result of channel name validation."""
+
+    valid: bool
+    warning: Optional[str] = None
+    suggestion: Optional[str] = None
+
+
+def validate_channel_name(channel: str) -> ChannelValidationResult:
+    """
+    Validate Centrifugo channel name and detect potential issues.
+
+    Common issues:
+    - Using `#` for namespace separator (should use `:`)
+    - User-limited channels without proper JWT token setup
+
+    Args:
+        channel: Channel name to validate
+
+    Returns:
+        Validation result with warnings and suggestions
+
+    Examples:
+        >>> # ❌ Bad: might be interpreted as user-limited channel
+        >>> result = validate_channel_name('terminal#session#abc123')
+        >>> print(result.warning)
+        Channel "terminal#session#abc123" uses '#' separator...
+
+        >>> # ✅ Good: proper namespace separator
+        >>> result = validate_channel_name('terminal:session:abc123')
+        >>> print(result.valid)
+        True
+    """
+    # Count # symbols
+    hash_count = channel.count("#")
+
+    if hash_count >= 2:
+        # Pattern: namespace#something#something
+        # This might be interpreted as user-limited channel: namespace#user_id#channel
+        parts = channel.split("#")
+        namespace, possible_user_id, *rest = parts
+
+        # Check if second part looks like a user ID (numeric)
+        is_numeric_user_id = possible_user_id.isdigit()
+
+        if not is_numeric_user_id and possible_user_id:
+            # Non-numeric second part after # - likely a mistake
+            suggestion = channel.replace("#", ":")
+
+            return ChannelValidationResult(
+                valid=False,
+                warning=(
+                    f'Channel "{channel}" uses \'#\' separator which Centrifugo interprets as '
+                    f'user-limited channel boundary. The part "{possible_user_id}" will be treated '
+                    f"as user_id, which may cause permission errors if not in JWT token."
+                ),
+                suggestion=f"Use ':' for namespace separation: \"{suggestion}\"",
+            )
+
+        if is_numeric_user_id:
+            return ChannelValidationResult(
+                valid=True,
+                warning=(
+                    f'Channel "{channel}" appears to be a user-limited channel '
+                    f'(user_id: {possible_user_id}). Make sure your JWT token\'s "sub" '
+                    f'field matches "{possible_user_id}".'
+                ),
+            )
+
+    # Single # is okay for user-limited channels like "user#123"
+    if hash_count == 1:
+        namespace, user_id_part = channel.split("#")
+        if user_id_part and not user_id_part.isdigit() and user_id_part != "*":
+            # Non-numeric user_id (not a wildcard) - might be a mistake
+            suggestion = channel.replace("#", ":")
+            return ChannelValidationResult(
+                valid=False,
+                warning=(
+                    f'Channel "{channel}" uses \'#\' but "{user_id_part}" doesn\'t look '
+                    f"like a user_id. This might cause permission issues."
+                ),
+                suggestion=f"Consider using ':' instead: \"{suggestion}\"",
+            )
+
+    return ChannelValidationResult(valid=True)
+
+
+def log_channel_warnings(channel: str, *, raise_warning: bool = False) -> None:
+    """
+    Log channel validation warnings (development only).
+
+    Args:
+        channel: Channel name to validate
+        raise_warning: If True, raises Python warning instead of logging
+
+    Examples:
+        >>> # Log to logger
+        >>> log_channel_warnings('terminal#session#abc123')
+
+        >>> # Raise Python warning (will be visible in tests)
+        >>> log_channel_warnings('terminal#session#abc123', raise_warning=True)
+    """
+    # Skip in production (if DEBUG=False)
+    try:
+        from django.conf import settings
+
+        if not settings.DEBUG:
+            return
+    except (ImportError, AttributeError):
+        # Django not configured or DEBUG not set - continue anyway
+        pass
+
+    result = validate_channel_name(channel)
+
+    if not result.valid and result.warning:
+        message = f"[Centrifugo Channel Warning] {result.warning}"
+        if result.suggestion:
+            message += f"\n💡 Suggestion: {result.suggestion}"
+
+        if raise_warning:
+            warnings.warn(message, UserWarning, stacklevel=2)
+        else:
+            logger.warning(message)
+
+    elif result.warning:
+        # Valid but has informational warning
+        if raise_warning:
+            warnings.warn(f"[Centrifugo] {result.warning}", UserWarning, stacklevel=2)
+        else:
+            logger.info(f"[Centrifugo] {result.warning}")
+
+
+__all__ = [
+    "ChannelValidationResult",
+    "validate_channel_name",
+    "log_channel_warnings",
+]

packages/django_cfg/apps/integrations/centrifugo/services/client/direct_client.py

@@ -165,6 +165,10 @@ async def publish(
             ...     data={"status": "running", "timestamp": "2025-11-05T09:00:00Z"}
             ... )
         """
+        # Validate channel name and log warnings (development only)
+        from ..channel_validator import log_channel_warnings
+        log_channel_warnings(channel)
+
         message_id = str(uuid4())
         start_time = time.time()
 

packages/django_cfg/core/builders/security_builder.py

@@ -219,10 +219,11 @@ def _get_prod_allowed_hosts(self, domain_hosts: List[str]) -> List[str]:
 
         In production we're strict, but need to allow:
         - Public domains (security_domains)
+        - Docker internal service names (auto-detected)
         - Docker health checks (internal IPs, if needed)
 
         Problem: If we allow all IPs - insecure!
-        Solution: Allow only private IP ranges (RFC 1918).
+        Solution: Allow only private IP ranges (RFC 1918) + internal service names.
 
         Args:
             domain_hosts: List of normalized domain hostnames
@@ -234,6 +235,10 @@ def _get_prod_allowed_hosts(self, domain_hosts: List[str]) -> List[str]:
 
         # Check if running in Docker
         if self._is_running_in_docker():
+            # Auto-detect internal service names from configuration
+            internal_services = self._get_internal_service_names()
+            allowed_hosts.extend(internal_services)
+
             # Allow Docker/Kubernetes health checks
             # Use regex for private IPs (RFC 1918)
             allowed_hosts.extend([
@@ -250,6 +255,127 @@ def _get_prod_allowed_hosts(self, domain_hosts: List[str]) -> List[str]:
 
         return allowed_hosts
 
+    def _get_internal_service_names(self) -> List[str]:
+        """
+        Auto-detect Docker internal service names from configuration.
+
+        Extracts service hostnames from internal URLs:
+        - Centrifugo API URL (centrifugo_api_url)
+        - gRPC internal URL (grpc.internal_url if available)
+        - Any other internal service URLs
+
+        Example:
+            centrifugo_api_url = "http://djangocfg-centrifugo:8000/api"
+            → Extracts: "djangocfg-centrifugo"
+
+        Returns:
+            List of internal service hostnames (without port)
+        """
+        service_names = []
+
+        # Extract from Centrifugo config
+        if hasattr(self.config, 'centrifugo') and self.config.centrifugo:
+            centrifugo_cfg = self.config.centrifugo
+
+            # Extract from centrifugo_api_url (for Django → Centrifugo publishing)
+            if hasattr(centrifugo_cfg, 'centrifugo_api_url'):
+                api_url = centrifugo_cfg.centrifugo_api_url
+                hostname = self._extract_hostname_from_url(api_url)
+                if hostname and self._is_internal_service_name(hostname):
+                    service_names.append(hostname)
+
+        # Extract from gRPC config
+        if hasattr(self.config, 'grpc') and self.config.grpc:
+            grpc_cfg = self.config.grpc
+
+            # Extract from internal_url (for container-to-container gRPC)
+            if hasattr(grpc_cfg, 'internal_url'):
+                internal_url = grpc_cfg.internal_url
+                # Handle formats: "djangocfg-grpc:50051" or "http://djangocfg-grpc:50051"
+                hostname = self._extract_hostname_from_url(internal_url, allow_no_protocol=True)
+                if hostname and self._is_internal_service_name(hostname):
+                    service_names.append(hostname)
+
+        # Deduplicate while preserving order
+        return list(dict.fromkeys(service_names))
+
+    def _extract_hostname_from_url(self, url: str, allow_no_protocol: bool = False) -> str:
+        """
+        Extract hostname from URL.
+
+        Args:
+            url: URL string (e.g., "http://djangocfg-api:8000/path" or "djangocfg-grpc:50051")
+            allow_no_protocol: Allow URLs without protocol (for gRPC addresses)
+
+        Returns:
+            Hostname without port (e.g., "djangocfg-api")
+        """
+        if not url:
+            return ""
+
+        try:
+            # Handle URLs without protocol (e.g., "djangocfg-grpc:50051")
+            if allow_no_protocol and not url.startswith(("http://", "https://")):
+                # Split by : to get hostname:port
+                hostname_port = url.split('/')[0]  # Remove path if any
+                hostname = hostname_port.split(':')[0]  # Remove port
+                return hostname.strip()
+
+            # Standard URL parsing
+            parsed = urlparse(url if url.startswith(("http://", "https://")) else f"http://{url}")
+            hostname = parsed.hostname or parsed.netloc.split(':')[0]
+            return hostname.strip() if hostname else ""
+        except Exception:
+            return ""
+
+    def _is_internal_service_name(self, hostname: str) -> bool:
+        """
+        Check if hostname is an internal Docker/Kubernetes service name.
+
+        Internal service names typically:
+        - Don't contain dots (unlike domains: example.com)
+        - Are not IPs (172.x.x.x, 192.168.x.x)
+        - Are not localhost/127.0.0.1
+
+        Examples:
+            "djangocfg-api" → True (internal service)
+            "djangocfg-grpc" → True (internal service)
+            "api.example.com" → False (external domain)
+            "192.168.1.10" → False (IP address)
+            "localhost" → False (localhost)
+
+        Args:
+            hostname: Hostname to check
+
+        Returns:
+            True if internal service name, False otherwise
+        """
+        if not hostname:
+            return False
+
+        hostname = hostname.lower().strip()
+
+        # Exclude localhost
+        if hostname in ('localhost', '127.0.0.1'):
+            return False
+
+        # Exclude IP addresses (simple check)
+        if hostname.replace('.', '').isdigit():
+            return False
+
+        # Exclude domains with dots (external domains like api.example.com)
+        # Keep Kubernetes service names like service.namespace.svc.cluster.local
+        if '.' in hostname:
+            # Allow .cluster.local and .svc (Kubernetes)
+            if hostname.endswith(('.cluster.local', '.svc')):
+                return True
+            # Otherwise it's an external domain
+            return False
+
+        # If no dots and not excluded - it's an internal service name
+        # Examples: djangocfg-api, djangocfg-grpc, postgres, redis
+        return True
+
     def _is_running_in_docker(self) -> bool:
         """
         Detect if application is running in Docker.

packages/django_cfg/models/api/grpc/config.py

@@ -376,6 +376,11 @@ class GRPCConfig(BaseConfig):
         description="Public URL for clients (e.g., 'grpc.djangocfg.com:443'). If None, auto-generated from api_url",
     )
 
+    internal_url: Optional[str] = Field(
+        default=None,
+        description="Internal Docker/Kubernetes URL for container-to-container communication (e.g., 'djangocfg-grpc:50051' or 'localhost:50051')",
+    )
+
     enable_reflection: Optional[bool] = Field(
         default=None,
         description="Enable server reflection for grpcurl/tools. If None, uses server.enable_reflection (True by default)",

packages/django_cfg/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "django-cfg"
-version = "1.5.115"
+version = "1.5.117"
 description = "Modern Django framework with type-safe Pydantic v2 configuration, Next.js admin integration, real-time WebSockets, and 8 enterprise apps. Replace settings.py with validated models, 90% less code. Production-ready with AI agents, auto-generated TypeScript clients, and zero-config features."
 readme = "README.md"
 keywords = [ "django", "configuration", "pydantic", "settings", "type-safety", "pydantic-settings", "django-environ", "startup-validation", "ide-autocomplete", "nextjs-admin", "react-admin", "websocket", "centrifugo", "real-time", "typescript-generation", "ai-agents", "enterprise-django", "django-settings", "type-safe-config", "modern-django",]