utils.py

import datetime
import ipaddress
import json
import os

from cryptography import x509
from cryptography.x509.oid import NameOID
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import rsa

DEFAULT_CERT_FILE = "server.crt"
DEFAULT_KEY_FILE = "server.key"

# Wire markers. XML-RPC's <int> is limited to 32 bits and <double> loses
# precision on large integers, so ints/floats travel as tagged strings instead.
INT_PREFIX = "__INT__"
FLOAT_PREFIX = "__FLOAT__"
# Escape marker: a genuine string that happens to look like a wire marker is
# prefixed with this so it does not decode back into a number.
STR_PREFIX = "__STR__"
_MARKER_PREFIXES = (INT_PREFIX, FLOAT_PREFIX, STR_PREFIX)

# Reused across calls; validating with iterencode avoids materialising the
# whole JSON string just to find out whether a value is serialisable.
_JSON_ENCODER = json.JSONEncoder()


def _write_secret(path, data):
    """Write bytes to path with owner-only permissions (0600)."""
    fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
    with os.fdopen(fd, "wb") as f:
        f.write(data)
    # Tighten permissions on a pre-existing file, which os.open leaves alone.
    os.chmod(path, 0o600)


def generate_self_signed_cert(cert_file=DEFAULT_CERT_FILE, key_file=DEFAULT_KEY_FILE):
    """Generate a self-signed certificate for HTTPS

    The certificate is for local development only: it is self-signed and valid
    for localhost/127.0.0.1, so clients must either trust it explicitly or
    disable verification.
    """

    # Generate private key
    private_key = rsa.generate_private_key(
        public_exponent=65537,
        key_size=4096,
    )

    # Create certificate
    subject = issuer = x509.Name([
        x509.NameAttribute(NameOID.COUNTRY_NAME, "US"),
        x509.NameAttribute(NameOID.STATE_OR_PROVINCE_NAME, "CA"),
        x509.NameAttribute(NameOID.LOCALITY_NAME, "San Francisco"),
        x509.NameAttribute(NameOID.ORGANIZATION_NAME, "Test Organization"),
        x509.NameAttribute(NameOID.COMMON_NAME, "localhost"),
    ])

    now = datetime.datetime.now(datetime.timezone.utc)

    cert = x509.CertificateBuilder().subject_name(
        subject
    ).issuer_name(
        issuer
    ).public_key(
        private_key.public_key()
    ).serial_number(
        x509.random_serial_number()
    ).not_valid_before(
        now
    ).not_valid_after(
        now + datetime.timedelta(days=365)
    ).add_extension(
        x509.SubjectAlternativeName([
            x509.DNSName("localhost"),
            x509.IPAddress(ipaddress.ip_address("127.0.0.1")),
        ]),
        critical=False,
    ).sign(private_key, hashes.SHA256())

    # Write certificate and key to files
    with open(cert_file, "wb") as f:
        f.write(cert.public_bytes(serialization.Encoding.PEM))

    # The private key must not be world-readable.
    _write_secret(key_file, private_key.private_bytes(
        encoding=serialization.Encoding.PEM,
        format=serialization.PrivateFormat.PKCS8,
        encryption_algorithm=serialization.NoEncryption()
    ))

    print(f"Self-signed certificate generated: {cert_file}, {key_file}")


def convert_value_for_xmlrpc(value):
    """Convert int/float values to string with type markers for XML-RPC transmission"""
    if isinstance(value, bool):
        return value  # Keep booleans as-is, check before int since bool is subclass of int
    elif isinstance(value, int):
        return f"{INT_PREFIX}{value}"
    elif isinstance(value, float):
        return f"{FLOAT_PREFIX}{value}"
    elif isinstance(value, str):
        # Escape strings that would otherwise be decoded as a numeric marker.
        if value.startswith(_MARKER_PREFIXES):
            return f"{STR_PREFIX}{value}"
        return value
    elif isinstance(value, (list, tuple)):
        return [convert_value_for_xmlrpc(item) for item in value]
    elif isinstance(value, dict):
        return {k: convert_value_for_xmlrpc(v) for k, v in value.items()}
    else:
        return value


def convert_value_from_xmlrpc(value):
    """Convert string representations back to int/float, leaving original strings unchanged"""
    if isinstance(value, str):
        if value.startswith(STR_PREFIX):
            # Escaped string: strip exactly one marker prefix.
            return value[len(STR_PREFIX):]
        elif value.startswith(INT_PREFIX):
            try:
                return int(value[len(INT_PREFIX):])
            except ValueError:
                return value  # Return unchanged if conversion fails
        elif value.startswith(FLOAT_PREFIX):
            try:
                return float(value[len(FLOAT_PREFIX):])
            except ValueError:
                return value  # Return unchanged if conversion fails
        else:
            return value  # Keep original strings as-is
    elif isinstance(value, (list, tuple)):
        return type(value)(convert_value_from_xmlrpc(item) for item in value)
    elif isinstance(value, dict):
        return {k: convert_value_from_xmlrpc(v) for k, v in value.items()}
    else:
        return value


class Data:
    """Utility class for passing args and kwargs to XML-RPC functions"""

    def _check_json_serializable(self, obj, name="argument"):
        """Check if an object is JSON serializable, raise exception if not."""
        try:
            for _ in _JSON_ENCODER.iterencode(obj):
                pass
        except (TypeError, ValueError) as e:
            raise ValueError(f"Non-JSON serializable {name}: {type(obj).__name__} - {str(e)}")

    def __init__(self, *args, response_code=None, error=None, result=None, **kwargs):
        # Check JSON serializability of args and kwargs before processing
        for i, arg in enumerate(args):
            self._check_json_serializable(arg, f"argument at position {i}")

        for key, value in kwargs.items():
            self._check_json_serializable(value, f"keyword argument '{key}'")

        # Convert int/float values to strings for XML-RPC transmission
        self.args = [convert_value_for_xmlrpc(arg) for arg in args]
        self.kwargs = {k: convert_value_for_xmlrpc(v) for k, v in kwargs.items()}
        # Timezone-aware: a naive local timestamp is ambiguous across hosts.
        self.timestamp = datetime.datetime.now(datetime.timezone.utc).isoformat()
        self.response_code = response_code if response_code is not None else 0
        self.error = error
        # Convert result if it's int/float
        self.result = convert_value_for_xmlrpc(result) if result is not None else result
        # Mark this as a Data object for XML-RPC serialization
        self._is_data_object = True

    def get_args(self):
        """Return the args as a tuple, decoded back to their original types"""
        return tuple(convert_value_from_xmlrpc(arg) for arg in self.args)

    def get_kwargs(self):
        """Return the kwargs as a dictionary, decoded back to their original types"""
        return {k: convert_value_from_xmlrpc(v) for k, v in self.kwargs.items()}

    def get_result(self):
        """Return the result value"""
        return convert_value_from_xmlrpc(self.result)

    def _get_dict_data(self):
        """Get dictionary representation without None values"""
        data = {
            'args': self.args,
            'kwargs': self.kwargs,
            'timestamp': self.timestamp,
            'response_code': self.response_code,
            'error': self.error,
            'result': self.result,
            '_is_data_object': True
        }
        # Remove any None values to avoid XML-RPC marshalling issues
        return {k: v for k, v in data.items() if v is not None}

    @property
    def __dict__(self):
        """Make the object appear as a dictionary to XML-RPC"""
        return self._get_dict_data()

    def __getstate__(self):
        """Automatically serialize for XML-RPC transmission"""
        return self._get_dict_data()

    def items(self):
        """Make the object iterable like a dictionary for XML-RPC marshalling"""
        return self._get_dict_data().items()

    def keys(self):
        """Dictionary-like interface for XML-RPC"""
        return self._get_dict_data().keys()

    def values(self):
        """Dictionary-like interface for XML-RPC"""
        return self._get_dict_data().values()

    def __getitem__(self, key):
        """Dictionary-like access for XML-RPC"""
        return self._get_dict_data()[key]

    def __repr__(self):
        return f"Data(timestamp={self.timestamp}, response_code={self.response_code}, error={self.error}, result={self.result}, args={self.args}, kwargs={self.kwargs})"