core/services/s3_service.py

"""
S3 storage service for backup operations.

Supports any S3-compatible provider: AWS, MinIO, DigitalOcean Spaces, Backblaze B2, etc.
"""

import ipaddress
import logging
import socket
from typing import Tuple
from urllib.parse import urlparse

from core.models import GlobalSettings
from core.services.encryption_service import EncryptionService, EncryptionError

logger = logging.getLogger(__name__)


def is_safe_endpoint_url(url: str) -> tuple[bool, str]:
    """
    Validate that an S3 endpoint URL doesn't point to internal/private resources.

    Blocks:
    - Private IP ranges (10.x, 172.16-31.x, 192.168.x)
    - Loopback addresses (127.x, localhost)
    - Link-local addresses (169.254.x - AWS metadata endpoint)
    - Internal hostnames

    Args:
        url: The endpoint URL to validate

    Returns:
        Tuple of (is_safe: bool, error_message: str)
    """
    if not url:
        return True, ""

    try:
        parsed = urlparse(url)
        hostname = parsed.hostname

        if not hostname:
            return False, "Invalid URL: no hostname found"

        # Block localhost variations
        if hostname.lower() in ("localhost", "localhost.localdomain"):
            return False, "Internal endpoints are not allowed (localhost)"

        # Try to resolve hostname to IP and check if it's private
        try:
            # Get all IP addresses for the hostname
            addr_info = socket.getaddrinfo(hostname, None)
            for _, _, _, _, sockaddr in addr_info:
                ip_str = sockaddr[0]
                ip = ipaddress.ip_address(ip_str)

                # Block private, loopback, and link-local addresses
                if ip.is_private:
                    return False, f"Private IP addresses are not allowed ({ip_str})"
                if ip.is_loopback:
                    return False, f"Loopback addresses are not allowed ({ip_str})"
                if ip.is_link_local:
                    return False, f"Link-local addresses are not allowed ({ip_str})"
                if ip.is_reserved:
                    return False, f"Reserved addresses are not allowed ({ip_str})"

        except socket.gaierror:
            # Hostname doesn't resolve - could be intentional for custom DNS
            # Allow it but log a warning
            logger.warning(f"S3 endpoint hostname does not resolve: {hostname}")

        return True, ""

    except Exception as e:
        return False, f"Invalid endpoint URL: {e}"


class S3ServiceError(Exception):
    """Raised when S3 operations fail."""

    pass


class S3Service:
    """
    Provides S3 storage operations for backup functionality.

    Supports any S3-compatible provider: AWS, MinIO, DigitalOcean Spaces, etc.
    """

    @classmethod
    def get_client(cls):
        """
        Create and return a configured boto3 S3 client.

        Returns:
            boto3 S3 client configured with current settings

        Raises:
            S3ServiceError: If S3 is not configured or credentials invalid
        """
        try:
            import boto3
            from botocore.config import Config
        except ImportError:
            raise S3ServiceError(
                "boto3 is not installed. Install it with: pip install boto3"
            )

        settings = GlobalSettings.get_settings()

        if not settings.s3_bucket_name:
            raise S3ServiceError("S3 bucket name is not configured")

        if not settings.s3_access_key_encrypted or not settings.s3_secret_key_encrypted:
            raise S3ServiceError("S3 credentials are not configured")

        # Decrypt credentials
        try:
            access_key = EncryptionService.decrypt(settings.s3_access_key_encrypted)
            secret_key = EncryptionService.decrypt(settings.s3_secret_key_encrypted)
        except EncryptionError as e:
            raise S3ServiceError(f"Failed to decrypt S3 credentials: {e}")

        # Build client config
        config = Config(
            signature_version="s3v4",
            s3={"addressing_style": "path" if settings.s3_path_style else "auto"},
        )

        client_kwargs = {
            "service_name": "s3",
            "aws_access_key_id": access_key,
            "aws_secret_access_key": secret_key,
            "config": config,
            "use_ssl": settings.s3_use_ssl,
        }

        if settings.s3_endpoint_url:
            # Validate endpoint URL to prevent SSRF attacks
            is_safe, error_msg = is_safe_endpoint_url(settings.s3_endpoint_url)
            if not is_safe:
                raise S3ServiceError(f"Invalid S3 endpoint: {error_msg}")
            client_kwargs["endpoint_url"] = settings.s3_endpoint_url

        if settings.s3_region:
            client_kwargs["region_name"] = settings.s3_region

        return boto3.client(**client_kwargs)

    @classmethod
    def test_connection(cls) -> Tuple[bool, str]:
        """
        Test the S3 connection by attempting to access the bucket.

        Returns:
            Tuple of (success: bool, message: str)
        """
        from django.utils import timezone

        settings = GlobalSettings.get_settings()

        try:
            client = cls.get_client()

            # Try to head the bucket (checks existence and permissions)
            client.head_bucket(Bucket=settings.s3_bucket_name)

            # Update last tested timestamp
            settings.s3_last_tested_at = timezone.now()
            settings.save(update_fields=["s3_last_tested_at"])

            return True, f"Successfully connected to bucket '{settings.s3_bucket_name}'"

        except S3ServiceError as e:
            return False, str(e)
        except Exception as e:
            error_msg = str(e)
            # Parse common boto3 errors for better messages
            if "NoSuchBucket" in error_msg:
                return False, f"Bucket '{settings.s3_bucket_name}' does not exist"
            elif "AccessDenied" in error_msg or "403" in error_msg:
                return (
                    False,
                    "Access denied. Check your credentials and bucket permissions",
                )
            elif "InvalidAccessKeyId" in error_msg:
                return False, "Invalid access key ID"
            elif "SignatureDoesNotMatch" in error_msg:
                return False, "Invalid secret access key"
            elif "EndpointConnectionError" in error_msg or "ConnectTimeoutError" in error_msg:
                return (
                    False,
                    "Cannot connect to endpoint. Check URL and network connectivity",
                )
            elif "InvalidEndpoint" in error_msg:
                return False, "Invalid endpoint URL format"
            else:
                logger.exception("S3 connection test failed")
                return False, f"Connection failed: {error_msg}"

    @classmethod
    def test_connection_with_credentials(
        cls,
        bucket_name: str,
        access_key: str,
        secret_key: str,
        endpoint_url: str = "",
        region: str = "us-east-1",
        use_ssl: bool = True,
        path_style: bool = False,
    ) -> Tuple[bool, str]:
        """
        Test S3 connection with provided credentials (without saving).

        This is used to validate credentials before saving settings.

        Args:
            bucket_name: S3 bucket name
            access_key: AWS access key ID
            secret_key: AWS secret access key
            endpoint_url: Custom endpoint URL (empty for AWS)
            region: AWS region
            use_ssl: Whether to use SSL
            path_style: Use path-style addressing

        Returns:
            Tuple of (success: bool, message: str)
        """
        # Validate required fields
        if not bucket_name:
            return False, "Bucket name is required"
        if not access_key:
            return False, "Access key is required"
        if not secret_key:
            return False, "Secret key is required"

        try:
            import boto3
            from botocore.config import Config
        except ImportError:
            return False, "boto3 is not installed. Install it with: pip install boto3"

        try:
            # Build client config
            config = Config(
                signature_version="s3v4",
                s3={"addressing_style": "path" if path_style else "auto"},
            )

            client_kwargs = {
                "service_name": "s3",
                "aws_access_key_id": access_key,
                "aws_secret_access_key": secret_key,
                "config": config,
                "use_ssl": use_ssl,
            }

            if endpoint_url:
                # Validate endpoint URL to prevent SSRF attacks
                is_safe, error_msg = is_safe_endpoint_url(endpoint_url)
                if not is_safe:
                    return False, f"Invalid S3 endpoint: {error_msg}"
                client_kwargs["endpoint_url"] = endpoint_url

            if region:
                client_kwargs["region_name"] = region

            client = boto3.client(**client_kwargs)

            # Try to head the bucket (checks existence and permissions)
            client.head_bucket(Bucket=bucket_name)

            return True, f"Successfully connected to bucket '{bucket_name}'"

        except Exception as e:
            error_msg = str(e)
            # Parse common boto3 errors for better messages
            if "NoSuchBucket" in error_msg:
                return False, f"Bucket '{bucket_name}' does not exist"
            elif "AccessDenied" in error_msg or "403" in error_msg:
                return (
                    False,
                    "Access denied. Check your credentials and bucket permissions",
                )
            elif "InvalidAccessKeyId" in error_msg:
                return False, "Invalid access key ID"
            elif "SignatureDoesNotMatch" in error_msg:
                return False, "Invalid secret access key"
            elif "EndpointConnectionError" in error_msg or "ConnectTimeoutError" in error_msg:
                return (
                    False,
                    "Cannot connect to endpoint. Check URL and network connectivity",
                )
            elif "InvalidEndpoint" in error_msg:
                return False, "Invalid endpoint URL format"
            else:
                logger.exception("S3 connection test failed")
                return False, f"Connection failed: {error_msg}"

    @classmethod
    def is_configured(cls) -> bool:
        """Check if S3 is properly configured (has required fields)."""
        settings = GlobalSettings.get_settings()
        return bool(
            settings.s3_bucket_name
            and settings.s3_access_key_encrypted
            and settings.s3_secret_key_encrypted
        )

    @classmethod
    def get_status(cls) -> dict:
        """
        Get S3 configuration status for UI display.

        Returns:
            Dict with status information
        """
        settings = GlobalSettings.get_settings()

        return {
            "enabled": settings.s3_enabled,
            "configured": cls.is_configured(),
            "bucket": settings.s3_bucket_name or None,
            "endpoint": settings.s3_endpoint_url or "AWS S3 (default)",
            "region": settings.s3_region or "us-east-1",
            "use_ssl": settings.s3_use_ssl,
            "path_style": settings.s3_path_style,
            "last_tested": settings.s3_last_tested_at,
        }

    @classmethod
    def upload_file(cls, file_bytes: bytes, key: str) -> dict:
        """
        Upload a file to S3.

        Args:
            file_bytes: File content as bytes
            key: S3 object key (path)

        Returns:
            dict: Upload result with 'success', 'key', 'size', optional 'error'
        """
        settings = GlobalSettings.get_settings()

        try:
            client = cls.get_client()
            client.put_object(
                Bucket=settings.s3_bucket_name,
                Key=key,
                Body=file_bytes,
                ContentType="application/gzip",
            )

            return {
                "success": True,
                "key": key,
                "size": len(file_bytes),
            }
        except S3ServiceError as e:
            logger.error(f"S3 upload failed for key {key}: {e}")
            return {
                "success": False,
                "key": key,
                "error": str(e),
            }
        except Exception as e:
            logger.exception(f"S3 upload failed for key {key}")
            return {
                "success": False,
                "key": key,
                "error": str(e),
            }

    @classmethod
    def list_files(cls, prefix: str = "") -> list[dict]:
        """
        List files in S3 bucket with optional prefix.

        Args:
            prefix: Key prefix to filter results

        Returns:
            list: List of dicts with 'key', 'size', 'last_modified'
        """
        settings = GlobalSettings.get_settings()

        try:
            client = cls.get_client()
            response = client.list_objects_v2(
                Bucket=settings.s3_bucket_name,
                Prefix=prefix,
            )

            files = []
            for obj in response.get("Contents", []):
                files.append({
                    "key": obj["Key"],
                    "size": obj["Size"],
                    "last_modified": obj["LastModified"],
                })

            return files
        except S3ServiceError as e:
            logger.error(f"S3 list failed for prefix {prefix}: {e}")
            return []
        except Exception as e:
            logger.exception(f"S3 list failed for prefix {prefix}")
            return []

    @classmethod
    def delete_file(cls, key: str) -> bool:
        """
        Delete a file from S3.

        Args:
            key: S3 object key to delete

        Returns:
            bool: True if deleted successfully
        """
        settings = GlobalSettings.get_settings()

        try:
            client = cls.get_client()
            client.delete_object(
                Bucket=settings.s3_bucket_name,
                Key=key,
            )
            return True
        except Exception as e:
            logger.exception(f"S3 delete failed for key {key}")
            return False

    @classmethod
    def delete_files(cls, keys: list[str]) -> int:
        """
        Delete multiple files from S3.

        Args:
            keys: List of S3 object keys to delete

        Returns:
            int: Number of files deleted
        """
        settings = GlobalSettings.get_settings()

        if not keys:
            return 0

        try:
            client = cls.get_client()
            response = client.delete_objects(
                Bucket=settings.s3_bucket_name,
                Delete={
                    "Objects": [{"Key": key} for key in keys],
                },
            )
            return len(response.get("Deleted", []))
        except Exception as e:
            logger.exception("S3 bulk delete failed")
            return 0

    @classmethod
    def generate_backup_key(cls, timestamp=None) -> str:
        """
        Generate a unique S3 key for a backup file.

        Args:
            timestamp: Optional datetime, defaults to now

        Returns:
            str: S3 key like "pyrunner-backups/backup_20240315_143022.json.gz"
        """
        from django.utils import timezone

        settings = GlobalSettings.get_settings()

        if timestamp is None:
            timestamp = timezone.now()

        prefix = settings.s3_backup_prefix.rstrip("/")
        filename = f"backup_{timestamp.strftime('%Y%m%d_%H%M%S')}.json.gz"

        return f"{prefix}/{filename}" if prefix else filename