"""Common classes"""
import ipaddress
import logging
import re
from typing import TYPE_CHECKING, Any, Optional, Union
import yaml
from django.apps import apps
from django.core.exceptions import AppRegistryNotReady
from django.utils.deprecation import MiddlewareMixin
from smarter.common.conf import smarter_settings
from smarter.common.exceptions import SmarterValueError
from smarter.common.helpers.console_helpers import (
formatted_text,
formatted_text_green,
formatted_text_red,
)
from smarter.common.utils import (
is_authenticated_request,
)
from smarter.common.utils import mask_string as util_mask_string
from smarter.common.utils import (
smarter_build_absolute_uri as utils_smarter_build_absolute_uri,
)
from smarter.lib import json
from smarter.lib.logging import WaffleSwitchedLoggerWrapper
if TYPE_CHECKING:
from django.http import HttpRequest
MOCK_REGEX = re.compile(r"<MagicMock|<Mock|mock\\.MagicMock|mock\\.Mock", re.IGNORECASE)
# guard against Sphinx doc build circular import errors
mixin_logging_is_active: bool = False
if apps.ready:
try:
# this resolves an import issue in collect static assets where Django apps are not yet importable
# pylint: disable=import-outside-toplevel,C0412
from smarter.lib.django import waffle
from smarter.lib.django.waffle import SmarterWaffleSwitches
mixin_logging_is_active = waffle.switch_is_active(SmarterWaffleSwitches.REQUEST_MIXIN_LOGGING)
# pylint: disable=broad-except
except (AppRegistryNotReady, ImportError):
pass
# pylint: disable=W0613
def should_log(level):
"""Check if logging should be done based on the waffle switch."""
return mixin_logging_is_active
base_logger = logging.getLogger(__name__)
logger = WaffleSwitchedLoggerWrapper(base_logger, should_log)
class Singleton(type):
"""
A metaclass for creating singleton classes.
usage:
class MyClass(metaclass=Singleton):
"""
_instances = {}
def __call__(cls, *args, **kwargs):
if cls not in cls._instances:
cls._instances[cls] = super().__call__(*args, **kwargs)
return cls._instances[cls]
[docs]
class SmarterHelperMixin:
"""
A generic mixin providing helper functions for Smarter classes.
This mixin offers utility methods and properties commonly needed
across Smarter classes, such as standardized class name formatting,
URL amnesty lists, JSON serialization, and data conversion.
"""
[docs]
def __init__(self, *args, **kwargs):
logger.debug("%s.__init__() - initializing with args=%s, kwargs=%s", self.formatted_class_name, args, kwargs)
@property
def formatted_class_name(self) -> str:
"""
Returns the class name formatted for logging.
:return: The formatted class name as a string.
:rtype: str
"""
return formatted_text(self.__class__.__name__)
@property
def unformatted_class_name(self) -> str:
"""
Returns the raw class name without formatting.
:return: The unformatted class name as a string.
:rtype: str
This is useful for logging or serialization where the plain class name is needed.
"""
return self.__class__.__name__
@property
def formatted_state_ready(self) -> str:
"""
Returns the readiness state formatted for logging.
:return: The formatted readiness state as a string.
:rtype: str
"""
return formatted_text_green("READY")
@property
def formatted_state_not_ready(self) -> str:
"""
Returns the not-ready state formatted for logging.
:return: The formatted not-ready state as a string.
:rtype: str
"""
return formatted_text_red("NOT_READY")
@property
def ready(self) -> bool:
"""
Indicates whether the object is ready for use. This is a placeholder
that should be overridden in subclasses.
:return: True if ready, False otherwise.
:rtype: bool
"""
return True
@property
def amnesty_urls(self) -> list[str]:
"""
Returns a list of URLs that are exempt from certain checks.
This is a placeholder and should be overridden in subclasses.
:return: List of URL path strings that are exempt.
:rtype: list[str]
"""
return ["readiness", "healthz", "favicon.ico", "robots.txt", "sitemap.xml"]
[docs]
def smarter_build_absolute_uri(self, request: "HttpRequest") -> Optional[str]:
"""
Attempts to get the absolute URI from a request object.
This utility function tries to retrieve the request URL from any valid
child class of :class:`django.http.HttpRequest`. It is especially useful
in unit tests or scenarios where the request object may not implement
``build_absolute_uri()``.
:param request: The request object.
:type request: Optional[HttpRequest]
:return: The absolute request URL.
:rtype: Optional[str]
:raises SmarterValueError: If the URI cannot be built from the request.
"""
return utils_smarter_build_absolute_uri(request)
[docs]
def mask_string(
self, string: Optional[str] = "", mask_char: str = "*", mask_length: int = 4, string_length: int = 8
) -> str:
"""
Masks a string for secure logging.
This utility function masks all but the last `unmasked_chars` characters
of the input string, replacing them with asterisks. It is useful for
logging sensitive information like API keys or passwords.
:param string: The string to be masked.
:type string: str
:param mask_char: The character used for masking.
:type mask_char: str
:param mask_length: The number of characters to mask.
:type mask_length: int
:param string_length: The length of the string to consider for masking.
:type string_length: int
:return: The masked string.
:rtype: str
"""
return util_mask_string(
string=string, mask_char=mask_char, mask_length=mask_length, string_length=string_length # type: ignore
)
[docs]
def data_to_dict(self, data: Union[dict, str]) -> dict:
"""
Converts data to a dictionary, handling different types of input.
This method accepts either a dictionary or a string. If a string is provided,
it will attempt to parse it as JSON first, and if that fails, as YAML.
If parsing fails or the data type is unsupported, a SmarterValueError is raised.
:param data: The data to convert, either a dict or a JSON/YAML string.
:type data: dict or str
:return: The data as a dictionary.
:rtype: dict
:raises SmarterValueError: If the data cannot be converted to a dictionary.
"""
if isinstance(data, dict):
return data
elif isinstance(data, str):
try:
return json.loads(data)
except json.JSONDecodeError:
try:
return yaml.safe_load(data)
except yaml.YAMLError as yaml_error:
raise SmarterValueError("String data is neither valid JSON nor YAML.") from yaml_error
else:
raise SmarterValueError("Unsupported data type for conversion to dict.")
[docs]
def sorted_dict(self, data: dict) -> dict:
"""
Returns a new dictionary with keys sorted.
:param data: The dictionary to sort.
:type data: dict
:return: A new dictionary with sorted keys.
:rtype: dict
"""
return {k: data[k] for k in sorted(data.keys())}
[docs]
class SmarterMiddlewareMixin(MiddlewareMixin, SmarterHelperMixin):
"""A mixin for middleware classes with helper functions.
This mixin provides utilities for extracting client IP addresses,
checking authentication indicators, and other middleware-related helpers.
Inherits from both Django's :class:`MiddlewareMixin` and :class:`SmarterHelperMixin`.
"""
def __call__(self, request):
logger.debug(
"%s.__call__() - processing request: %s",
self.formatted_class_name,
self.smarter_build_absolute_uri(request),
)
return super().__call__(request)
[docs]
def get_client_ip(self, request) -> Optional[str]:
"""
Get client IP address from request.
This method attempts to determine the original client IP address,
accounting for proxies, load balancers, and CDNs. It checks common
headers set by proxies and falls back to ``REMOTE_ADDR``.
Notes
-----
- In AWS CLB → Kubernetes Nginx setups, the client IP flow is:
Client → CLB → Nginx Ingress → Django.
- CLB adds ``X-Forwarded-For`` with the original client IP.
- Nginx may add ``X-Real-IP`` or modify ``X-Forwarded-For``.
- Django sees ``REMOTE_ADDR`` as the Nginx IP (not the client IP).
- If using Cloudflare, it adds the ``CF-Connecting-IP`` header.
- Always validate IPs to avoid trusting spoofed headers.
:param request: The Django request object.
:type request: HttpRequest
:return: The detected client IP address, or None if not found.
:rtype: Optional[str]
"""
# First check X-Forwarded-For (most reliable for CLB)
# set by Nginx ingress controller and Traefik.
# Contains the original client IP and any proxy IPs.
forwarded_for = request.META.get("HTTP_X_FORWARDED_FOR")
if forwarded_for:
# X-Forwarded-For format: "client_ip, proxy1_ip, proxy2_ip"
# The leftmost IP is the original client IP
client_ip = forwarded_for.split(",")[0].strip()
# Validate it's not a private IP (load balancer/proxy IP)
if not self._is_private_ip(client_ip):
logger.debug(
"%s.get_client_ip() - Using X-Forwarded-For: %s",
self.formatted_class_name,
client_ip,
)
return client_ip
# Check X-Real-IP (set by Nginx ingress controller and Traefik)
real_ip = request.META.get("HTTP_X_REAL_IP")
if real_ip and not self._is_private_ip(real_ip.strip()):
logger.debug(
"%s.get_client_ip() - Using X-Real-IP: %s",
self.formatted_class_name,
real_ip.strip(),
)
return real_ip.strip()
# Check Cloudflare connecting IP if using Cloudflare
cf_ip = request.META.get("HTTP_CF_CONNECTING_IP")
if cf_ip and not self._is_private_ip(cf_ip.strip()):
logger.debug(
"%s.get_client_ip() - Using CF-Connecting-IP: %s",
self.formatted_class_name,
cf_ip.strip(),
)
return cf_ip.strip()
# Fallback to REMOTE_ADDR (will be load balancer IP in AWS)
remote_addr = request.META.get("REMOTE_ADDR", "127.0.0.1")
logger.debug(
"%s.get_client_ip() - Falling back to REMOTE_ADDR: %s",
self.formatted_class_name,
remote_addr,
)
if not self._is_private_ip(remote_addr):
logger.debug(
"%s.get_client_ip() - Using REMOTE_ADDR: %s",
self.formatted_class_name,
remote_addr,
)
return remote_addr
if request.path.replace("/", "") not in self.amnesty_urls and not smarter_settings.environment_is_local:
logger.warning(
"%s __call()__ - Could not determine client IP: %s, Meta: %s",
self.formatted_class_name,
self.smarter_build_absolute_uri(request=request),
request.META,
)
return None
def _is_private_ip(self, ip):
"""Check if IP is in private/internal ranges."""
try:
ip_obj = ipaddress.ip_address(ip)
return ip_obj.is_private or ip_obj.is_loopback or ip_obj.is_link_local
except ValueError as e:
# Regex to match MagicMock or mock object string representations
ip_str = str(ip)
if MOCK_REGEX.search(ip_str) or "Mock" in getattr(ip, "__class__", type(ip)).__name__:
logger.warning(
"%s._is_private_ip() - Mock object detected as IP: %s", self.formatted_class_name, ip_str
)
else:
logger.warning(
"%s._is_private_ip() - Invalid IP address: %s, error: %s", self.formatted_class_name, ip_str, e
)
return True
[docs]
def has_auth_indicators(self, request):
"""
Check if request has authentication indicators (cookies, headers, etc.).
This method inspects the request for common authentication signals,
such as session cookies, CSRF tokens, authorization headers, API keys,
or Django's built-in authentication.
:param request: The Django request object.
:type request: HttpRequest
:return: True if authentication indicators are present, False otherwise.
:rtype: bool
"""
# Check for Django session cookie
if request.COOKIES.get("sessionid"):
return True
# Check for CSRF token (indicates active session)
if request.COOKIES.get("csrftoken"):
return True
# Check for Authorization header
if request.META.get("HTTP_AUTHORIZATION"):
return True
# Check for API key header
if request.META.get("HTTP_X_API_KEY"):
return True
# Check if user is authenticated (Django built-in)
if is_authenticated_request(request):
return True
return False