"""Configuration loading and validation.""" from __future__ import annotations import json import re from dataclasses import dataclass from pathlib import Path from typing import List class ConfigError(Exception): """Configuration file load/validation failure.""" @dataclass class PolicyConfig: policy_type: str policy_name: str @dataclass class RoleConfig: configuration_name: str description: str session_duration: int policies: List[PolicyConfig] @dataclass class AppConfig: directory_id: str region: str roles: List[RoleConfig] _DIRECTORY_ID_PATTERN = re.compile(r"^d-[A-Za-z0-9]+$") _DEFAULT_REGION = "cn-shanghai" _ALLOWED_POLICY_TYPES = {"System", "Custom"} def _require_non_empty_str(value, field_name: str) -> str: if not isinstance(value, str) or not value.strip(): raise ConfigError(f"{field_name} must be a non-empty string") return value.strip() def _resolve_region(config_region, region_override: str = None) -> str: if region_override is not None: return _require_non_empty_str(region_override, "Region override") if config_region is None: return _DEFAULT_REGION return _require_non_empty_str(config_region, "Region") def load_config(config_path: str, region_override: str = None) -> AppConfig: """Load and validate config file into typed dataclasses.""" path = Path(config_path) if not path.exists(): raise ConfigError(f"Config file not found: {config_path}") try: raw = json.loads(path.read_text(encoding="utf-8")) except json.JSONDecodeError as exc: raise ConfigError(f"Invalid JSON in config file: {exc}") from exc except OSError as exc: raise ConfigError(f"Unable to read config file: {exc}") from exc if not isinstance(raw, dict): raise ConfigError("Config root must be a JSON object") directory_id = _require_non_empty_str(raw.get("DirectoryId"), "DirectoryId") if not _DIRECTORY_ID_PATTERN.match(directory_id): raise ConfigError("DirectoryId must match pattern d-*") roles_raw = raw.get("Roles") if not isinstance(roles_raw, list) or not roles_raw: raise ConfigError("Roles must be a non-empty array") roles: List[RoleConfig] = [] for role_idx, role_raw in enumerate(roles_raw): if not isinstance(role_raw, dict): raise ConfigError(f"Roles[{role_idx}] must be an object") configuration_name = _require_non_empty_str( role_raw.get("ConfigurationName"), f"Roles[{role_idx}].ConfigurationName", ) description = _require_non_empty_str( role_raw.get("Description"), f"Roles[{role_idx}].Description", ) session_duration = role_raw.get("SessionDuration") if not isinstance(session_duration, int): raise ConfigError(f"Roles[{role_idx}].SessionDuration must be an integer") if session_duration < 900 or session_duration > 43200: raise ConfigError( f"Roles[{role_idx}].SessionDuration must be in range [900, 43200]" ) policies_raw = role_raw.get("Policies") if not isinstance(policies_raw, list) or not policies_raw: raise ConfigError(f"Roles[{role_idx}].Policies must be a non-empty array") policies: List[PolicyConfig] = [] for policy_idx, policy_raw in enumerate(policies_raw): if not isinstance(policy_raw, dict): raise ConfigError( f"Roles[{role_idx}].Policies[{policy_idx}] must be an object" ) policy_type = _require_non_empty_str( policy_raw.get("PolicyType"), f"Roles[{role_idx}].Policies[{policy_idx}].PolicyType", ) if policy_type not in _ALLOWED_POLICY_TYPES: raise ConfigError( f"Roles[{role_idx}].Policies[{policy_idx}].PolicyType must be one of " f"{sorted(_ALLOWED_POLICY_TYPES)}" ) policy_name = _require_non_empty_str( policy_raw.get("PolicyName"), f"Roles[{role_idx}].Policies[{policy_idx}].PolicyName", ) policies.append(PolicyConfig(policy_type=policy_type, policy_name=policy_name)) roles.append( RoleConfig( configuration_name=configuration_name, description=description, session_duration=session_duration, policies=policies, ) ) region = _resolve_region(raw.get("Region"), region_override) return AppConfig(directory_id=directory_id, region=region, roles=roles)