config.py

from __future__ import annotations

import shlex
from dataclasses import dataclass
from pathlib import Path
from typing import Callable, Optional
import ipaddress
import re

from timecapsulesmb.core.net import extract_host, ipv4_literal, is_link_local_ipv4
from timecapsulesmb.core.paths import package_project_root, resolve_app_paths

REPO_ROOT = package_project_root()
ENV_PATH = REPO_ROOT / ".env"
MAX_DNS_LABEL_BYTES = 63
MAX_DNS_NAME_BYTES = 255
MAX_DNS_TXT_BYTES = 255
MAX_NETBIOS_NAME_BYTES = 15
MODEL_TXT_PREFIX = "model="
ADISK_DEFAULT_DISK_KEY = "dk2"
ADISK_MANAGED_DISK_ADVF = "0x82"
ADISK_DISK_UUID_EXAMPLE = "12345678-1234-1234-1234-123456789012"
ADISK_DISK_TXT_ADVF_PREFIX = "=adVF="
ADISK_DISK_TXT_ADVN_MID = ",adVN="
ADISK_DISK_TXT_SUFFIX = ",adVU="
MAX_SAMBA_USER_BYTES = 32
MANAGED_PAYLOAD_DIR_NAME = ".samba4"
DEFAULT_SAMBA_AUTH_USER = "admin"
DEFAULT_MDNS_DEVICE_MODEL = "TimeCapsule"


@dataclass(frozen=True)
class AirportDeviceIdentity:
    syap: str
    mdns_model: str
    display_name: str
    family: str
    compatibility_group: str


AIRPORT_DEVICE_IDENTITIES = (
    AirportDeviceIdentity("104", "AirPort5,104", "AirPort Extreme 1st generation", "airport_extreme", "netbsd4be"),
    AirportDeviceIdentity("105", "AirPort5,105", "AirPort Extreme 2nd generation", "airport_extreme", "netbsd4be"),
    AirportDeviceIdentity("106", "TimeCapsule6,106", "Time Capsule 1st generation", "time_capsule", "netbsd4be"),
    AirportDeviceIdentity("108", "AirPort5,108", "AirPort Extreme 3rd generation", "airport_extreme", "netbsd4le"),
    AirportDeviceIdentity("109", "TimeCapsule6,109", "Time Capsule 2nd generation", "time_capsule", "netbsd4be"),
    AirportDeviceIdentity("113", "TimeCapsule6,113", "Time Capsule 3rd generation", "time_capsule", "netbsd4le"),
    AirportDeviceIdentity("114", "AirPort5,114", "AirPort Extreme 4th generation", "airport_extreme", "netbsd4le"),
    AirportDeviceIdentity("116", "TimeCapsule6,116", "Time Capsule 4th generation", "time_capsule", "netbsd4le"),
    AirportDeviceIdentity("117", "AirPort5,117", "AirPort Extreme 5th generation", "airport_extreme", "netbsd4le"),
    AirportDeviceIdentity("119", "TimeCapsule8,119", "Time Capsule 5th generation", "time_capsule", "netbsd6"),
    AirportDeviceIdentity("120", "AirPort7,120", "AirPort Extreme 6th generation", "airport_extreme", "netbsd6"),
)
AIRPORT_IDENTITIES_BY_SYAP = {identity.syap: identity for identity in AIRPORT_DEVICE_IDENTITIES}
AIRPORT_IDENTITIES_BY_MODEL = {identity.mdns_model: identity for identity in AIRPORT_DEVICE_IDENTITIES}
VALID_AIRPORT_SYAP_CODES = frozenset(AIRPORT_IDENTITIES_BY_SYAP)
VALID_MDNS_DEVICE_MODELS = frozenset(
    {"TimeCapsule", "AirPort"} | {identity.mdns_model for identity in AIRPORT_DEVICE_IDENTITIES}
)
AIRPORT_SYAP_TO_MODEL = {
    identity.syap: identity.mdns_model
    for identity in AIRPORT_DEVICE_IDENTITIES
}
DEFAULT_SSH_TARGET_PLACEHOLDER = "root@192.168.x.x"


def airport_identity_from_values(values: dict[str, str]) -> AirportDeviceIdentity | None:
    syap = values.get("TC_AIRPORT_SYAP", "")
    model = values.get("TC_MDNS_DEVICE_MODEL", "")
    return AIRPORT_IDENTITIES_BY_SYAP.get(syap) or AIRPORT_IDENTITIES_BY_MODEL.get(model)


DEFAULTS = {
    "TC_HOST": DEFAULT_SSH_TARGET_PLACEHOLDER,
    "TC_SSH_OPTS": "-o HostKeyAlgorithms=+ssh-rsa -o PubkeyAcceptedAlgorithms=+ssh-rsa -o KexAlgorithms=+diffie-hellman-group14-sha1 -o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null",
    "TC_INTERNAL_SHARE_USE_DISK_ROOT": "false",
    "TC_ANY_PROTOCOL": "false",
}

ENV_FILE_KEYS = [
    "TC_HOST",
    "TC_PASSWORD",
    "TC_SSH_OPTS",
    "TC_INTERNAL_SHARE_USE_DISK_ROOT",
    "TC_ANY_PROTOCOL",
    "TC_CONFIGURE_ID",
]

CONFIG_HEADER = """# Local user/device configuration for TimeCapsuleSMB.
# Generated by tcapsule configure
"""


class ConfigError(Exception):
    """Base class for recoverable configuration failures."""


@dataclass(frozen=True, init=False)
class AppConfig:
    path: Path
    exists: bool
    file_values: dict[str, str]
    values: dict[str, str]

    def __init__(
        self,
        *,
        values: dict[str, str] | None = None,
        path: Path | None = None,
        exists: bool = True,
        file_values: dict[str, str] | None = None,
    ) -> None:
        resolved_values = dict(values or {})
        resolved_file_values = dict(file_values or {})
        object.__setattr__(self, "path", path or default_env_path())
        object.__setattr__(self, "exists", exists)
        object.__setattr__(self, "file_values", resolved_file_values)
        object.__setattr__(self, "values", resolved_values)

    @classmethod
    def from_values(
        cls,
        values: dict[str, str] | None = None,
        *,
        path: Path | None = None,
        exists: bool = True,
        file_values: dict[str, str] | None = None,
    ) -> "AppConfig":
        return cls(values=values, path=path, exists=exists, file_values=file_values)

    @classmethod
    def from_file(cls, path: Path | None = None, *, defaults: Optional[dict[str, str]] = None) -> "AppConfig":
        resolved_path = path or default_env_path()
        resolved_defaults = dict(DEFAULTS if defaults is None else defaults)
        exists = resolved_path.exists()
        file_values = parse_env_file(resolved_path) if exists else {}
        values = dict(resolved_defaults)
        values.update(file_values)
        return cls(values=values, path=resolved_path, exists=exists, file_values=file_values)

    @classmethod
    def missing(cls, path: Path | None = None) -> "AppConfig":
        return cls(values={}, path=path, exists=False, file_values={})

    def get(self, key: str, default: str = "") -> str:
        return self.values.get(key, default)

    def has_file_value(self, key: str) -> bool:
        return bool(self.file_values.get(key, ""))

    def has_value(self, key: str) -> bool:
        return bool(self.values.get(key, ""))

    def require(self, key: str, *, messagebefore: str = "", messageafter: str = "") -> str:
        value = self.get(key)
        if not value:
            raise ConfigError(f"{messagebefore}Missing required setting in {self.path}: {key}{messageafter}")
        return value


def airport_identity_from_config(config: AppConfig) -> AirportDeviceIdentity | None:
    return (
        AIRPORT_IDENTITIES_BY_SYAP.get(config.get("TC_AIRPORT_SYAP"))
        or AIRPORT_IDENTITIES_BY_MODEL.get(config.get("TC_MDNS_DEVICE_MODEL"))
    )


def airport_identity_from_model_or_syap(
    *,
    model: str | None = None,
    syap: str | None = None,
) -> AirportDeviceIdentity | None:
    return (
        AIRPORT_IDENTITIES_BY_SYAP.get(syap or "")
        or AIRPORT_IDENTITIES_BY_MODEL.get(model or "")
    )


def airport_family_display_name_from_identity(
    *,
    model: str | None = None,
    syap: str | None = None,
) -> str:
    identity = airport_identity_from_model_or_syap(model=model, syap=syap)
    family = identity.family if identity is not None else ""
    if family == "time_capsule" or model == "TimeCapsule":
        return "Time Capsule"
    if family == "airport_extreme" or model == "AirPort":
        return "AirPort Extreme"
    return "AirPort storage device"


def airport_exact_display_name_from_identity(
    *,
    model: str | None = None,
    syap: str | None = None,
) -> str:
    identity = airport_identity_from_model_or_syap(model=model, syap=syap)
    if identity is not None:
        return identity.display_name
    return airport_family_display_name_from_identity(model=model, syap=syap)


def airport_family_display_name_from_config(config: AppConfig) -> str:
    model = config.get("TC_MDNS_DEVICE_MODEL")
    identity = airport_identity_from_config(config)
    family = identity.family if identity is not None else ""
    if family == "time_capsule" or model == "TimeCapsule":
        return "Time Capsule"
    if family == "airport_extreme" or model == "AirPort":
        return "AirPort Extreme"
    return "AirPort storage device"


def airport_exact_display_name_from_config(config: AppConfig) -> str:
    identity = airport_identity_from_config(config)
    if identity is not None:
        return identity.display_name
    return airport_family_display_name_from_config(config)


@dataclass(frozen=True)
class ConfigIssue:
    kind: str
    key: str | None
    message: str
    path: Path

    def format_for_cli(self, *, command_name: str | None = None) -> str:
        if self.kind == "missing_file":
            lines = [f"Missing required configuration file: {self.path}"]
        elif self.kind == "missing_key" and self.key:
            lines = [f"Missing required setting in {self.path}: {self.key}"]
        elif self.key:
            lines = [f"{self.key} is invalid in {self.path}. Run the `configure` command again.", self.message]
        else:
            lines = [self.message]
        if command_name:
            lines.append(f"Please run the `configure` command before running `{command_name}`.")
        return "\n".join(lines)


class ConfigValidationError(ConfigError):
    def __init__(self, issues: list[ConfigIssue], *, command_name: str | None = None) -> None:
        self.issues = issues
        self.command_name = command_name
        if issues:
            message = issues[0].format_for_cli(command_name=command_name)
        else:
            message = "Configuration validation failed."
        super().__init__(message)


def parse_env_value(raw_value: str) -> str:
    value = raw_value.strip()
    if not value:
        return ""
    try:
        tokens = shlex.split(value)
        # A single parsed token means the env value was one scalar, possibly
        # quoted. Multi-token values such as TC_SSH_OPTS must remain intact and
        # are interpreted later by the transport layer.
        if len(tokens) == 1:
            return tokens[0]
        return value
    except ValueError:
        return value.strip("'\"")


def parse_env_file(path: Path) -> dict[str, str]:
    values: dict[str, str] = {}
    if not path.exists():
        return values
    for raw_line in path.read_text().splitlines():
        line = raw_line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        key, value = line.split("=", 1)
        values[key.strip()] = parse_env_value(value)
    return values


def default_env_path() -> Path:
    return resolve_app_paths().config_path


def load_app_config(path: Path | None = None, *, defaults: Optional[dict[str, str]] = None) -> AppConfig:
    return AppConfig.from_file(path, defaults=defaults)


def shell_quote(value: str) -> str:
    return shlex.quote(value)


def _contains_invalid_control_character(value: str) -> bool:
    return any(ord(ch) < 0x20 or ord(ch) == 0x7F for ch in value)


def _contains_whitespace(value: str) -> bool:
    return any(ch.isspace() for ch in value)


def _has_only_safe_chars(value: str, pattern: str) -> bool:
    return re.fullmatch(pattern, value) is not None


def build_mdns_device_model_txt(value: str) -> Optional[str]:
    txt = MODEL_TXT_PREFIX + value
    if len(txt.encode("utf-8")) > MAX_DNS_TXT_BYTES:
        return None
    return txt


def build_adisk_share_txt(value: str) -> Optional[str]:
    txt = (
        f"{ADISK_DEFAULT_DISK_KEY}{ADISK_DISK_TXT_ADVF_PREFIX}{ADISK_MANAGED_DISK_ADVF}"
        f"{ADISK_DISK_TXT_ADVN_MID}{value}"
        f"{ADISK_DISK_TXT_SUFFIX}{ADISK_DISK_UUID_EXAMPLE}"
    )
    if len(txt.encode("utf-8")) > MAX_DNS_TXT_BYTES:
        return None
    return txt


def validate_mdns_device_model(value: str, field_name: str) -> Optional[str]:
    if not value:
        return f"{field_name} cannot be blank."
    if value not in VALID_MDNS_DEVICE_MODELS:
        return f"{field_name} is not a supported AirPort storage device model."
    if build_mdns_device_model_txt(value) is None:
        return f"{field_name} must be 249 bytes or fewer."
    if _contains_invalid_control_character(value):
        return f"{field_name} contains an invalid control character."
    return None


def validate_mdns_instance_name(value: str, field_name: str) -> Optional[str]:
    if not value:
        return f"{field_name} cannot be blank."
    if len(value.encode("utf-8")) > MAX_DNS_LABEL_BYTES:
        return f"{field_name} must be {MAX_DNS_LABEL_BYTES} bytes or fewer."
    if "." in value:
        return f"{field_name} must not contain dots."
    if _contains_invalid_control_character(value):
        return f"{field_name} contains an invalid control character."
    return None


def validate_mdns_host_label(value: str, field_name: str) -> Optional[str]:
    if not value:
        return f"{field_name} cannot be blank."
    if len(value.encode("utf-8")) > MAX_DNS_LABEL_BYTES:
        return f"{field_name} must be {MAX_DNS_LABEL_BYTES} bytes or fewer."
    try:
        ipaddress.ip_address(value)
        return f"{field_name} must be a single DNS label, not an IP address."
    except ValueError:
        pass
    if "." in value:
        return f"{field_name} must not contain dots."
    if value.startswith("-") or value.endswith("-"):
        return f"{field_name} must not start or end with a hyphen."
    if not _has_only_safe_chars(value, r"[A-Za-z0-9-]+"):
        return f"{field_name} may contain only letters, numbers, and hyphens."
    return None


def validate_netbios_name(value: str, field_name: str) -> Optional[str]:
    if not value:
        return f"{field_name} cannot be blank."
    if len(value.encode("utf-8")) > MAX_NETBIOS_NAME_BYTES:
        return f"{field_name} must be {MAX_NETBIOS_NAME_BYTES} bytes or fewer."
    if _contains_invalid_control_character(value):
        return f"{field_name} contains an invalid control character."
    if not _has_only_safe_chars(value, r"[A-Za-z0-9_-]+"):
        return f"{field_name} may contain only letters, numbers, underscores, and hyphens."
    return None


def validate_samba_user(value: str, field_name: str) -> Optional[str]:
    if not value:
        return f"{field_name} cannot be blank."
    if len(value.encode("utf-8")) > MAX_SAMBA_USER_BYTES:
        return f"{field_name} must be {MAX_SAMBA_USER_BYTES} bytes or fewer."
    if _contains_invalid_control_character(value):
        return f"{field_name} contains an invalid control character."
    if _contains_whitespace(value):
        return f"{field_name} must not contain whitespace."
    if not _has_only_safe_chars(value, r"[A-Za-z0-9._-]+"):
        return f"{field_name} may contain only letters, numbers, dots, underscores, and hyphens."
    return None


def validate_payload_dir_name(value: str, field_name: str) -> Optional[str]:
    if not value:
        return f"{field_name} cannot be blank."
    if value in {".", ".."}:
        return f"{field_name} must not be . or ..."
    if "/" in value or "\\" in value:
        return f"{field_name} must be a single directory name, not a path."
    if value.startswith("-"):
        return f"{field_name} must not start with a hyphen."
    if _contains_invalid_control_character(value):
        return f"{field_name} contains an invalid control character."
    if not _has_only_safe_chars(value, r"[A-Za-z0-9._-]+"):
        return f"{field_name} may contain only letters, numbers, dots, underscores, and hyphens."
    return None


def validate_net_iface(value: str, field_name: str) -> Optional[str]:
    if not value:
        return f"{field_name} cannot be blank."
    if _contains_invalid_control_character(value):
        return f"{field_name} contains an invalid control character."
    if _contains_whitespace(value):
        return f"{field_name} must not contain whitespace."
    if not _has_only_safe_chars(value, r"[A-Za-z0-9._:-]+"):
        return f"{field_name} may contain only letters, numbers, dots, underscores, colons, and hyphens."
    return None


def validate_ssh_target(value: str, field_name: str) -> Optional[str]:
    if not value:
        return f"{field_name} cannot be blank."
    if _contains_invalid_control_character(value):
        return f"{field_name} contains an invalid control character."
    if _contains_whitespace(value):
        return f"{field_name} must not contain whitespace."
    if "@" not in value:
        return f"{field_name} must include a username, like {DEFAULT_SSH_TARGET_PLACEHOLDER}"
    user, host = value.split("@", 1)
    if not user:
        return f"{field_name} must include a username before @."
    if not host:
        return f"{field_name} must include a host after @."
    if host.lower() == "192.168.x.x":
        return (
            f"{field_name} IP address is invalid. "
            "Replace 192.168.x.x with the device's actual IP address."
        )
    if not _has_only_safe_chars(user, r"[A-Za-z0-9._-]+"):
        return f"{field_name} username may contain only letters, numbers, dots, underscores, and hyphens."
    if host.startswith("-"):
        return f"{field_name} host must not start with a hyphen."
    host_ip = ipv4_literal(host)
    if host_ip is not None and is_link_local_ipv4(host_ip):
        return (
            f"{field_name} host must not be a 169.254.x.x link-local address. "
            "Use the device's LAN IP or a hostname that resolves to its LAN IP; "
            "169.254.x.x is only suitable for temporary SSH recovery."
        )
    return None


def parse_bool(value: str) -> bool:
    return value.strip().lower() == "true"


def validate_bool(value: str, field_name: str) -> Optional[str]:
    if value == "":
        return None
    if value.strip().lower() not in {"true", "false"}:
        return f"{field_name} must be true or false."
    return None


def validate_airport_syap(value: str, field_name: str) -> Optional[str]:
    if not value:
        return f"{field_name} cannot be blank."
    if not value.isdigit():
        return f"{field_name} must contain only digits."
    if value not in VALID_AIRPORT_SYAP_CODES:
        return "The configured syAP is invalid."
    return None


def infer_mdns_device_model_from_airport_syap(syap: str) -> Optional[str]:
    return AIRPORT_SYAP_TO_MODEL.get(syap)


def validate_mdns_device_model_matches_syap(syap: str, device_model: str) -> Optional[str]:
    expected_model = infer_mdns_device_model_from_airport_syap(syap)
    if expected_model is None:
        return None
    if device_model != expected_model:
        return (f'TC_MDNS_DEVICE_MODEL "{device_model}" must match the '
                f'configured syAP expected value "{expected_model}".')
    return None


CONFIG_VALIDATORS: dict[str, Callable[[str, str], Optional[str]]] = {
    "TC_HOST": validate_ssh_target,
    "TC_SAMBA_USER": validate_samba_user,
    "TC_PAYLOAD_DIR_NAME": validate_payload_dir_name,
    "TC_AIRPORT_SYAP": validate_airport_syap,
    "TC_MDNS_DEVICE_MODEL": validate_mdns_device_model,
    "TC_INTERNAL_SHARE_USE_DISK_ROOT": validate_bool,
    "TC_ANY_PROTOCOL": validate_bool,
}


@dataclass(frozen=True)
class ConfigProfile:
    required_file_values: tuple[str, ...] = ()
    required_values: tuple[str, ...] = ()
    validated_keys: tuple[str, ...] = ()
    require_env_file: bool = True
    cross_check_syap_model: bool = False


CONFIGURE_VALIDATED_KEYS = (
    "TC_INTERNAL_SHARE_USE_DISK_ROOT",
    "TC_ANY_PROTOCOL",
)
MANAGED_VALIDATED_KEYS = (
    "TC_HOST",
    "TC_INTERNAL_SHARE_USE_DISK_ROOT",
    "TC_ANY_PROTOCOL",
)
MANAGED_REQUIRED_FILE_KEYS = (
    "TC_HOST",
)
FLASH_REQUIRED_FILE_KEYS = (
    "TC_HOST",
    "TC_PASSWORD",
)
FLASH_VALIDATED_KEYS = (
    "TC_HOST",
)

CONFIG_PROFILES: dict[str, ConfigProfile] = {
    "configure": ConfigProfile(
        validated_keys=CONFIGURE_VALIDATED_KEYS,
        require_env_file=False,
    ),
    "deploy": ConfigProfile(
        required_file_values=MANAGED_REQUIRED_FILE_KEYS,
        validated_keys=MANAGED_VALIDATED_KEYS,
    ),
    "activate": ConfigProfile(
        required_file_values=MANAGED_REQUIRED_FILE_KEYS,
        validated_keys=MANAGED_VALIDATED_KEYS,
    ),
    "doctor": ConfigProfile(
        required_file_values=(*MANAGED_REQUIRED_FILE_KEYS, "TC_PASSWORD"),
        validated_keys=MANAGED_VALIDATED_KEYS,
    ),
    "uninstall": ConfigProfile(
        required_file_values=("TC_HOST",),
        validated_keys=("TC_HOST",),
    ),
    "fsck": ConfigProfile(
        required_file_values=("TC_HOST",),
        validated_keys=("TC_HOST",),
    ),
    "set_ssh": ConfigProfile(
        required_file_values=("TC_HOST", "TC_PASSWORD"),
        validated_keys=("TC_HOST",),
    ),
    "flash": ConfigProfile(
        required_file_values=FLASH_REQUIRED_FILE_KEYS,
        validated_keys=FLASH_VALIDATED_KEYS,
    ),
    "repair_xattrs": ConfigProfile(
        required_values=("TC_HOST",),
        validated_keys=("TC_HOST",),
        require_env_file=False,
    ),
}
def validate_app_config(config: AppConfig, *, profile: str) -> list[ConfigIssue]:
    profile_config = CONFIG_PROFILES[profile]
    if profile_config.require_env_file and not config.exists:
        return [
            ConfigIssue(
                kind="missing_file",
                key=None,
                message=f"Missing required configuration file: {config.path}",
                path=config.path,
            )
        ]

    errors: list[ConfigIssue] = []
    missing_keys: set[str] = set()
    for key in profile_config.required_file_values:
        if not config.has_file_value(key):
            missing_keys.add(key)
            errors.append(ConfigIssue(
                kind="missing_key",
                key=key,
                message=f"Missing required setting in {config.path}: {key}",
                path=config.path,
            ))
    for key in profile_config.required_values:
        if not config.has_value(key):
            missing_keys.add(key)
            errors.append(ConfigIssue(
                kind="missing_key",
                key=key,
                message=f"Missing required setting in {config.path}: {key}",
                path=config.path,
            ))

    for key in profile_config.validated_keys:
        if key in missing_keys:
            continue
        validator = CONFIG_VALIDATORS.get(key)
        if validator is None:
            continue
        error = validator(config.get(key, ""), key)
        if error:
            errors.append(ConfigIssue(
                kind="invalid_value",
                key=key,
                message=error,
                path=config.path,
            ))
    if profile_config.cross_check_syap_model and "TC_AIRPORT_SYAP" not in missing_keys and "TC_MDNS_DEVICE_MODEL" not in missing_keys:
        syap_model_error = validate_mdns_device_model_matches_syap(
            config.get("TC_AIRPORT_SYAP", ""),
            config.get("TC_MDNS_DEVICE_MODEL", ""),
        )
        if syap_model_error:
            errors.append(ConfigIssue(
                kind="inconsistent_values",
                key="TC_MDNS_DEVICE_MODEL",
                message=syap_model_error,
                path=config.path,
            ))
    return errors


def require_valid_app_config(config: AppConfig, *, profile: str, command_name: str | None = None) -> None:
    errors = validate_app_config(config, profile=profile)
    if errors:
        raise ConfigValidationError(errors, command_name=command_name)


def render_env_text(values: dict[str, str]) -> str:
    lines = [CONFIG_HEADER.rstrip(), ""]
    for key in ENV_FILE_KEYS:
        rendered_value = values.get(key, DEFAULTS.get(key, ""))
        lines.append(f"{key}={shell_quote(rendered_value)}")
    lines.append("")
    return "\n".join(lines)


def write_env_file(path: Path, values: dict[str, str]) -> None:
    path.write_text(render_env_text(values))