Orchestrator/shared/AI-Orchestrator/scripts/pat_manager/registry.py

"""PAT Registry module for loading, validating, and persisting the PAT registry."""

from __future__ import annotations

import json
import re
from pathlib import Path

from .errors import RegistryValidationError
from .models import PatEntry

# Valid service names and their required renewal methods
VALID_SERVICES: dict[str, str] = {
    "GitLab": "auto",
    "Jira": "manual",
    "Confluence": "manual",
    "OrgMyLife": "manual",
}

# ISO 8601 date pattern (YYYY-MM-DD)
_ISO_DATE_PATTERN = re.compile(r"^\d{4}-\d{2}-\d{2}$")

# Maximum token name length
_MAX_TOKEN_NAME_LENGTH = 128


class PatRegistry:
    """Manages loading, validation, and persistence of the PAT registry.

    The registry is a JSON file with structure:
        {"version": "1.0", "tokens": [...]}

    Each token entry contains: service, token_name, expiry_date, renewal_method.
    """

    def load(self, path: Path) -> list[PatEntry]:
        """Load and validate all entries from a PAT registry JSON file.

        Args:
            path: Path to the registry JSON file.

        Returns:
            List of validated PatEntry objects.

        Raises:
            FileNotFoundError: If the registry file does not exist.
            json.JSONDecodeError: If the file contains invalid JSON.
            RegistryValidationError: If any entry fails validation.
        """
        with open(path, "r", encoding="utf-8") as f:
            data = json.load(f)

        if not isinstance(data, dict):
            raise RegistryValidationError(
                entry={},
                failed_fields=["root"],
            )

        tokens = data.get("tokens", [])
        if not isinstance(tokens, list):
            raise RegistryValidationError(
                entry={},
                failed_fields=["tokens"],
            )

        entries: list[PatEntry] = []
        seen: set[tuple[str, str]] = set()

        for raw_entry in tokens:
            entry = self.validate_entry(raw_entry)
            # Check for duplicates
            key = (entry.service, entry.token_name)
            if key in seen:
                raise RegistryValidationError(
                    entry=raw_entry,
                    failed_fields=["service", "token_name"],
                )
            seen.add(key)
            entries.append(entry)

        return entries

    def validate_entry(self, entry: dict) -> PatEntry:
        """Validate a single registry entry dictionary.

        Checks:
            - All required fields are present (service, token_name, renewal_method)
            - service is one of the supported values
            - token_name is non-empty and max 128 characters
            - expiry_date is null or valid ISO 8601 (YYYY-MM-DD)
            - renewal_method matches the service (GitLab -> "auto", others -> "manual")

        Args:
            entry: Raw dictionary representing a PAT entry.

        Returns:
            A validated PatEntry instance.

        Raises:
            RegistryValidationError: If validation fails, with specific failed fields.
        """
        if not isinstance(entry, dict):
            raise RegistryValidationError(
                entry=entry if isinstance(entry, dict) else {"_raw": str(entry)},
                failed_fields=["entry_type"],
            )

        failed_fields: list[str] = []

        # Check required fields presence
        required_fields = ["service", "token_name", "renewal_method"]
        for field in required_fields:
            if field not in entry:
                failed_fields.append(field)

        # If required fields are missing, raise immediately
        if failed_fields:
            raise RegistryValidationError(entry=entry, failed_fields=failed_fields)

        # Validate service
        service = entry["service"]
        if service not in VALID_SERVICES:
            failed_fields.append("service")

        # Validate token_name
        token_name = entry["token_name"]
        if (
            not isinstance(token_name, str)
            or len(token_name) == 0
            or len(token_name) > _MAX_TOKEN_NAME_LENGTH
        ):
            failed_fields.append("token_name")

        # Validate expiry_date (can be null/None or valid ISO 8601 YYYY-MM-DD)
        expiry_date = entry.get("expiry_date")
        if expiry_date is not None:
            if not isinstance(expiry_date, str) or not _is_valid_iso_date(expiry_date):
                failed_fields.append("expiry_date")

        # Validate renewal_method matches service
        renewal_method = entry["renewal_method"]
        if service in VALID_SERVICES:
            expected_method = VALID_SERVICES[service]
            if renewal_method != expected_method:
                failed_fields.append("renewal_method")
        elif renewal_method not in ("auto", "manual"):
            # If service is invalid, still check renewal_method is at least valid
            failed_fields.append("renewal_method")

        if failed_fields:
            raise RegistryValidationError(entry=entry, failed_fields=failed_fields)

        return PatEntry(
            service=service,
            token_name=token_name,
            expiry_date=expiry_date,
            renewal_method=renewal_method,
        )

    def save(self, path: Path, entries: list[PatEntry]) -> None:
        """Persist a list of PAT entries to a registry JSON file.

        Args:
            path: Path where the registry file should be written.
            entries: List of PatEntry objects to serialize.
        """
        data = {
            "version": "1.0",
            "tokens": [
                {
                    "service": entry.service,
                    "token_name": entry.token_name,
                    "expiry_date": entry.expiry_date,
                    "renewal_method": entry.renewal_method,
                }
                for entry in entries
            ],
        }

        path.parent.mkdir(parents=True, exist_ok=True)
        with open(path, "w", encoding="utf-8") as f:
            json.dump(data, f, indent=2)
            f.write("\n")

    def create_empty(self, path: Path) -> None:
        """Create a new empty registry file with valid JSON structure.

        Args:
            path: Path where the empty registry file should be created.
        """
        self.save(path, [])


def _is_valid_iso_date(date_str: str) -> bool:
    """Check if a string is a valid ISO 8601 date (YYYY-MM-DD).

    Validates both the format and that the date values are actually valid
    (e.g., rejects 2025-02-30).
    """
    if not _ISO_DATE_PATTERN.match(date_str):
        return False

    # Validate actual date values
    try:
        year, month, day = date_str.split("-")
        year_int, month_int, day_int = int(year), int(month), int(day)

        # Basic range checks
        if month_int < 1 or month_int > 12:
            return False
        if day_int < 1 or day_int > 31:
            return False

        # Use datetime for full validation (handles leap years, etc.)
        import datetime

        datetime.date(year_int, month_int, day_int)
        return True
    except (ValueError, OverflowError):
        return False