Config module¶

def load_config() -> dict[str, Any]:
    """
    Load the current configuration from the config file. Returns an empty
    dictionary if no configuration file exists.

    Returns
    -------
    dict[str, Any]
        The current configuration as a dictionary.
    """
    if not CONFIG_FILE.exists():
        return {}

    with CONFIG_FILE.open() as file:
        config = yaml.safe_load(file)

    return config if isinstance(config, dict) else {}

def save_config(config: dict[str, Any]) -> None:
    """
    Save the given configuration to the config file.

    Parameters
    ----------
    config : dict[str, Any]
        The configuration to save.
    """
    CONFIG_DIR.mkdir(parents=True, exist_ok=True)

    with CONFIG_FILE.open("w") as file:
        yaml.safe_dump(config, file)

    # Restrict read/write access to the owner.  Best-effort: silently ignored
    # on filesystems or platforms that don't support POSIX permissions.
    try:
        os.chmod(CONFIG_FILE, 0o600)
    except (OSError, PermissionError):
        pass

def load_sessions() -> dict[str, Any]:
    """
    Load the sessions configuration from ``~/.nextmv/sessions.yaml``.

    Returns an empty dictionary if the file does not exist.  The returned dict
    is keyed by endpoint hostname; each value is a mapping with at least
    ``oidc_discovery_url`` and ``client_id``.

    Returns
    -------
    dict[str, Any]
        The sessions configuration, or ``{}`` if the file is absent.
    """
    if not SESSIONS_FILE.exists():
        return {}

    with SESSIONS_FILE.open() as fh:
        data = yaml.safe_load(fh)

    return data if isinstance(data, dict) else {}

def save_sessions(sessions: dict[str, Any]) -> None:
    """
    Persist the sessions configuration to ``~/.nextmv/sessions.yaml``.

    Parameters
    ----------
    sessions : dict[str, Any]
        The full sessions mapping to write, keyed by endpoint hostname.
    """
    CONFIG_DIR.mkdir(parents=True, exist_ok=True)

    with SESSIONS_FILE.open("w") as fh:
        yaml.safe_dump(sessions, fh)

    try:
        os.chmod(SESSIONS_FILE, 0o600)
    except (OSError, PermissionError):
        pass

def get_endpoint_oidc_config(endpoint: str, sessions: dict[str, Any] | None = None) -> dict[str, str] | None:
    """
    Return the OIDC configuration (``oidc_discovery_url`` and ``client_id``)
    for *endpoint*.

    Lookup order:

    1. *sessions* (the caller-supplied dict, typically loaded from
       ``sessions.yaml``).
    2. The built-in fallback table :data:`_BUILTIN_OIDC` — covers
       ``api.cloud.nextmv.io`` so that existing users don't need to re-run
       ``nextmv configuration create``.

    Parameters
    ----------
    endpoint : str
        The endpoint hostname, e.g. ``"api.cloud.nextmv.io"``.  Leading
        ``https://`` or ``http://`` are stripped automatically.
    sessions : dict[str, Any] | None
        The sessions mapping loaded from ``sessions.yaml``.  Pass ``None`` (or
        omit) to skip the file lookup and rely solely on the built-in table.

    Returns
    -------
    dict[str, str] | None
        A dict with ``oidc_discovery_url`` and ``client_id`` keys, or ``None``
        if the endpoint is not known.
    """
    # Normalise the endpoint to a bare hostname.
    ep = _strip_scheme(endpoint)

    if sessions:
        entry = sessions.get(ep)
        if isinstance(entry, dict):
            url = entry.get(OIDC_DISCOVERY_URL_KEY)
            cid = entry.get(CLIENT_ID_KEY)
            if url and cid:
                return {OIDC_DISCOVERY_URL_KEY: _normalize_oidc_discovery_url(url), CLIENT_ID_KEY: cid}

    return _BUILTIN_OIDC.get(ep)

def non_profile_keys() -> set[str]:
    """
    Returns the set of top-level config keys that are not profile names.

    Returns
    -------
    set[str]
        The set of non-profile keys.
    """
    return {
        API_KEY_KEY,
        ENDPOINT_KEY,
        AUTH_TYPE_KEY,
        AUTH_SESSION_KEY,
        TEAM_ID_KEY,
        SYSTEM_CERTS_KEY,
        SSO_DOMAIN_KEY,
        DEFAULT_AUTH_SESSION,
    }

def get_auth_type(config: dict, profile: str | None) -> AuthType:
    """
    Returns the auth type for the given profile. Defaults to
    ``AuthType.API_KEY`` if the key is absent (backwards compatible).

    Parameters
    ----------
    config : dict
        The full configuration dictionary loaded from config.yaml.
    profile : str | None
        The profile name. If None, the default (top-level) profile is used.

    Returns
    -------
    AuthType
        Either ``AuthType.API_KEY`` or ``AuthType.PKCE``.
    """
    if profile is None:
        raw = config.get(AUTH_TYPE_KEY, AuthType.API_KEY)
    else:
        profile_data = config.get(profile, {})
        if not isinstance(profile_data, dict):
            return AuthType.API_KEY
        raw = profile_data.get(AUTH_TYPE_KEY, AuthType.API_KEY)
    try:
        return AuthType(raw)
    except ValueError:
        return AuthType.API_KEY

def get_auth_session(config: dict, profile: str | None) -> str:
    """
    Returns the auth session name to use for token storage for *profile*.

    When ``auth_session`` is explicitly set on the profile, that value is
    returned.  Otherwise the reserved ``DEFAULT_AUTH_SESSION`` (``"default"``)
    is returned, meaning that all profiles without an explicit session share a
    single set of tokens.  This is the expected UX for users who have only one
    Nextmv identity.

    Parameters
    ----------
    config : dict
        The full configuration dictionary loaded from config.yaml.
    profile : str | None
        The profile name.  If ``None``, the default (top-level) profile is
        used.

    Returns
    -------
    str
        The session name (never empty; falls back to ``DEFAULT_AUTH_SESSION``).
    """
    if profile is None:
        raw = config.get(AUTH_SESSION_KEY)
    else:
        profile_data = config.get(profile, {})
        raw = profile_data.get(AUTH_SESSION_KEY) if isinstance(profile_data, dict) else None

    if raw and isinstance(raw, str) and raw.strip():
        return raw.strip()
    return DEFAULT_AUTH_SESSION

def get_team_id(config: dict, profile: str | None) -> str | None:
    """
    Returns the team (organization) UUID stored in *profile*, or ``None`` if
    not set.

    Parameters
    ----------
    config : dict
        The full configuration dictionary loaded from config.yaml.
    profile : str | None
        The profile name.  If ``None``, the default (top-level) profile is
        used.

    Returns
    -------
    str | None
        The team UUID, or ``None`` when the key is absent or the profile is
        not a ``pkce`` profile.
    """
    if profile is None:
        raw = config.get(TEAM_ID_KEY)
    else:
        profile_data = config.get(profile, {})
        raw = profile_data.get(TEAM_ID_KEY) if isinstance(profile_data, dict) else None

    if raw and isinstance(raw, str) and raw.strip():
        return raw.strip()
    return None

def get_system_certs(config: dict, profile: str | None) -> bool:
    """
    Returns ``True`` if *profile* is configured to use the system certificate
    store for TLS connections (via the ``truststore`` package).

    Parameters
    ----------
    config : dict
        The full configuration dictionary loaded from config.yaml.
    profile : str | None
        The profile name.  If ``None``, the default (top-level) profile is
        used.

    Returns
    -------
    bool
        ``True`` when ``system_certs`` is set to a truthy value, ``False``
        otherwise.
    """
    if profile is None:
        raw = config.get(SYSTEM_CERTS_KEY, False)
    else:
        profile_data = config.get(profile, {})
        raw = profile_data.get(SYSTEM_CERTS_KEY, False) if isinstance(profile_data, dict) else False
    return bool(raw)

def get_sso_domain(config: dict, profile: str | None) -> str | None:
    """Return the SSO domain stored in *profile*, or ``None`` if not set."""
    if profile is None:
        raw = config.get(SSO_DOMAIN_KEY)
    else:
        profile_data = config.get(profile, {})
        raw = profile_data.get(SSO_DOMAIN_KEY) if isinstance(profile_data, dict) else None
    if raw and isinstance(raw, str) and raw.strip():
        return raw.strip()
    return None

def get_profile_endpoint(config: dict, profile: str | None) -> str:
    """
    Returns the endpoint hostname for *profile*.

    Parameters
    ----------
    config : dict
        The full configuration dictionary loaded from config.yaml.
    profile : str | None
        The profile name.  If ``None``, the default (top-level) profile is
        used.

    Returns
    -------
    str
        The endpoint hostname, falling back to ``DEFAULT_ENDPOINT`` when not
        set.
    """
    if profile is None:
        ep = config.get(ENDPOINT_KEY, DEFAULT_ENDPOINT)
    else:
        profile_data = config.get(profile, {})
        ep = profile_data.get(ENDPOINT_KEY, DEFAULT_ENDPOINT) if isinstance(profile_data, dict) else DEFAULT_ENDPOINT

    if ep is None:
        return DEFAULT_ENDPOINT
    return _strip_scheme(str(ep)) or DEFAULT_ENDPOINT

def list_pkce_profiles(config: dict) -> list[str | None]:
    """
    Returns a list of profile names (or ``None`` for the default profile)
    whose profile type is ``pkce``.

    Parameters
    ----------
    config : dict
        The full configuration dictionary loaded from config.yaml.

    Returns
    -------
    list[str | None]
        A list where each entry is either a named profile string or ``None``
        (representing the default profile).
    """
    result: list[str | None] = []
    if config.get(AUTH_TYPE_KEY) == AuthType.PKCE:
        result.append(None)
    skip = non_profile_keys()
    for key, value in config.items():
        if key in skip:
            continue
        if isinstance(value, dict) and value.get(AUTH_TYPE_KEY) == AuthType.PKCE:
            result.append(key)
    return result