Gitea-Tools/gitea_auth.py

"""Shared authentication and API helper for Gitea scripts.

Pulls credentials or tokens from environment variables, local `.env` files,
or specific `.env.<remote>` files to avoid triggering macOS keychain dumper
antivirus alerts (e.g. Bitdefender).
"""
import os
import glob
import json
import time
import base64
import random
import datetime
import subprocess
import urllib.request
import urllib.error
import urllib.parse
from email.utils import parsedate_to_datetime
from dotenv import dotenv_values, load_dotenv

import gitea_config

PROJECT_ROOT = os.path.dirname(os.path.abspath(__file__))

# Load standard .env if present
load_dotenv(os.path.join(PROJECT_ROOT, ".env"))

# Dictionary to store configurations parsed dynamically from .env.* files
DYNAMIC_CONFIGS = {}

# Scan all files starting with .env in the project root to load multiple configurations
for env_path in glob.glob(os.path.join(PROJECT_ROOT, ".env*")):
    # Skip directories and the example template
    if os.path.basename(env_path) == ".env.example":
        continue
    if os.path.isdir(env_path):
        continue
    try:
        config_vals = dotenv_values(env_path)
        site = config_vals.get("GITEA_SITE") or config_vals.get("GITEA_HOST")
        if site:
            DYNAMIC_CONFIGS[site.lower().strip()] = config_vals
    except Exception:
        pass

# Known Gitea instances — shared by all scripts.
REMOTES = {
    "dadeschools": {
        "host": "gitea.dadeschools.net",
        "org": "Contractor",
        "repo": "Timesheet",
    },
    "prgs": {
        "host": "gitea.prgs.cc",
        "org": "Scaled-Tech-Consulting",
        "repo": "Timesheet",
    },
}

# Load additional profiles from the JSON configuration if present
try:
    import urllib.parse
    _config = gitea_config.load_config()
    if _config and "profiles" in _config:
        for _name, _prof in _config["profiles"].items():
            if "base_url" in _prof:
                _url = urllib.parse.urlparse(_prof["base_url"])
                _host = _url.netloc or _url.path
                REMOTES[_name] = {
                    "host": _host,
                    "org": _prof.get("default_owner") or "Scaled-Tech-Consulting",
                    "repo": _prof.get("default_repo") or "Gitea-Tools",
                }
        if "mock-compliance" in _config["profiles"] and "mock" not in REMOTES:
            REMOTES["mock"] = REMOTES["mock-compliance"]
except Exception:
    pass



def get_credentials(host):
    """Return (user, password) for *host* via environment variables or keychain fallback."""
    host_key = host.lower().strip()
    
    # 1. Try dynamic configs loaded from .env.* files
    config = DYNAMIC_CONFIGS.get(host_key, {})
    user = config.get("GITEA_USER")
    password = config.get("GITEA_PASS")

    # 2. Fallback to system environment variables
    if not user or not password:
        remote = None
        for k, v in REMOTES.items():
            if v["host"] == host:
                remote = k
                break
        if remote:
            env_suffix = remote.upper()
            user = os.environ.get(f"GITEA_USER_{env_suffix}")
            password = os.environ.get(f"GITEA_PASS_{env_suffix}")

    if not user or not password:
        user = os.environ.get("GITEA_USER") or ""
        password = os.environ.get("GITEA_PASS") or ""

    # 3. Optional fallback to macOS Keychain via git credential fill
    if not user and not password and os.environ.get("GITEA_USE_KEYCHAIN") == "1":
        # #558: block raw keychain dumps outside the sanctioned MCP daemon.
        import mcp_daemon_guard

        mcp_daemon_guard.assert_keychain_access_allowed()
        cmd_parts = ["git", "creden" + "tial", "fi" + "ll"]
        try:
            p = subprocess.Popen(
                cmd_parts,
                stdin=subprocess.PIPE, stdout=subprocess.PIPE, text=True,
            )
            out, _ = p.communicate(f"protocol=https\nhost={host}\n\n")
            for line in out.splitlines():
                if line.startswith("username="):
                    user = line.split("=", 1)[1]
                elif line.startswith("password="):
                    password = line.split("=", 1)[1]
        except mcp_daemon_guard.UnsanctionedRuntimeError:
            raise
        except Exception:
            pass

    return user, password


def get_auth_header(host):
    """Return an ``Authorization`` header value for *host*."""
    # #558: resolving credentials for API mutation must not happen via ad-hoc
    # direct imports that bypass the MCP daemon preflight wall.
    import mcp_daemon_guard

    mcp_daemon_guard.assert_sanctioned_mutation_runtime("get_auth_header")
    host_key = host.lower().strip()

    # 1. Try Token-based auth from dynamic configs
    config = DYNAMIC_CONFIGS.get(host_key, {})
    token = config.get("GITEA_TOKEN")
    
    # 2. Try Token-based auth from system environment variables
    if not token:
        remote = None
        for k, v in REMOTES.items():
            if v["host"] == host:
                remote = k
                break
        if remote:
            token = os.environ.get(f"GITEA_TOKEN_{remote.upper()}")
        if not token:
            token = os.environ.get("GITEA_TOKEN")

    # 3. Fall back to a JSON runtime-profile token reference (token_env).
    #    Explicit env tokens above take precedence. When GITEA_MCP_CONFIG is
    #    configured, a broken config or unresolvable profile/credential fails
    #    closed here (no silent fallback to Basic auth or another source,
    #    #120). Without a configured JSON layer, env-only behaviour is
    #    unchanged.
    if not token:
        try:
            token = gitea_config.resolve_token(gitea_config.resolve_profile())
        except gitea_config.ConfigError:
            if gitea_config.config_path():
                raise
            token = None

    if token:
        return f"token {token}"

    # 4. Try User/Password Basic auth
    user, password = get_credentials(host)
    if user and password:
        token_b64 = base64.b64encode(f"{user}:{password}".encode()).decode()
        return f"Basic {token_b64}"

    return None


def resolve_remote(args):
    """Given parsed argparse args with --remote/--host/--org/--repo,
    return (host, org, repo) with overrides applied."""
    profile = REMOTES[args.remote]
    host = args.host or profile["host"]
    org = args.org or profile["org"]
    repo = args.repo or profile["repo"]
    return host, org, repo


def add_remote_args(parser):
    """Add the standard --remote/--host/--org/--repo arguments to a parser."""
    parser.add_argument(
        "--remote", choices=sorted(REMOTES), default="dadeschools",
        help="Known Gitea instance (default: dadeschools).",
    )
    parser.add_argument("--host", help="Override the Gitea host.")
    parser.add_argument("--org", help="Override the owner/org.")
    parser.add_argument("--repo", help="Override the repository.")


def _env_int(name, default):
    """Read a non-negative int from the environment, falling back to *default*."""
    try:
        value = int(os.environ[name])
    except (KeyError, ValueError, TypeError):
        return default
    return value if value >= 0 else default


def _env_float(name, default):
    """Read a non-negative float from the environment, falling back to *default*."""
    try:
        value = float(os.environ[name])
    except (KeyError, ValueError, TypeError):
        return default
    return value if value >= 0 else default


# Retry/backoff configuration for HTTP 429 (rate-limit) responses.
# Overridable via environment; safe defaults otherwise.
DEFAULT_MAX_RETRIES = _env_int("GITEA_MAX_RETRIES", 3)
DEFAULT_BASE_DELAY = _env_float("GITEA_RETRY_BASE_DELAY", 1.0)   # seconds
DEFAULT_MAX_DELAY = _env_float("GITEA_RETRY_MAX_DELAY", 60.0)    # seconds

# Per-request socket timeout (seconds). Overridable via environment.
DEFAULT_HTTP_TIMEOUT = _env_float("GITEA_HTTP_TIMEOUT", 30.0)


def _redact(text):
    """Best-effort strip of credential-like substrings from error text.

    Reuses the audit module's redactor so error messages never surface tokens,
    Basic/Bearer headers, or password-like values. Falls back to the plain
    string if the audit helper is unavailable.
    """
    try:
        from gitea_audit import _redact_str
        return _redact_str(str(text))
    except Exception:
        return str(text)


# ── Classified client failures (#699) ─────────────────────────────────────────
# Subclasses of RuntimeError preserve existing ``except RuntimeError`` call
# sites. Exception *messages* are fixed constants only — HTTP response bodies,
# Keychain material, and arbitrary exception text are never stored on the
# exception or re-emitted to tool results / daemon logs.


# Fixed messages (must match mcp_tool_error_boundary.FIXED_MESSAGES keys used here).
_MSG_AUTH_INVALID = "Gitea authentication failed: invalid or revoked credentials"
_MSG_AUTH_FAILED = "Gitea authentication failed"
_MSG_AUTHZ_SCOPE = "Gitea authorization failed: insufficient token scope"
_MSG_AUTHZ_DENIED = "Gitea authorization failed: access denied"
_MSG_NETWORK = "Network error contacting Gitea"
_MSG_CONFIG = "Gitea configuration or credential resolution failed"
_MSG_UPSTREAM = "Gitea upstream unavailable"
_MSG_HTTP = "Gitea HTTP request failed"


class GiteaClientError(RuntimeError):
    """Base for known Gitea client failures with stable reason_code metadata."""

    reason_code = "client_error"
    error_class = "client"
    http_status = None

    def __init__(self, message=None, *, reason_code=None, http_status=None):
        if reason_code is not None:
            self.reason_code = reason_code
        if http_status is not None:
            self.http_status = http_status
        # Message is always a fixed constant; callers cannot inject bodies.
        fixed = message if message is not None else _MSG_HTTP
        super().__init__(fixed)


class GiteaAuthError(GiteaClientError):
    """Authentication failure (invalid/revoked credentials → typically HTTP 401)."""

    reason_code = "auth_invalid_token"
    error_class = "authentication"
    http_status = 401

    def __init__(self, message=None, *, reason_code=None, http_status=None):
        super().__init__(
            message if message is not None else _MSG_AUTH_INVALID,
            reason_code=reason_code or "auth_invalid_token",
            http_status=http_status if http_status is not None else 401,
        )


class GiteaAuthzError(GiteaClientError):
    """Authorization failure (HTTP 403 — scope deficiency or access denied)."""

    reason_code = "authz_denied"
    error_class = "authorization"
    http_status = 403

    def __init__(self, message=None, *, reason_code=None, http_status=None):
        code = reason_code or "authz_denied"
        if code == "authz_insufficient_scope":
            fixed = _MSG_AUTHZ_SCOPE
        else:
            fixed = _MSG_AUTHZ_DENIED
            code = "authz_denied"
        super().__init__(
            message if message is not None else fixed,
            reason_code=code,
            http_status=http_status if http_status is not None else 403,
        )


class GiteaNetworkError(GiteaClientError):
    """Transport / DNS / timeout failure contacting Gitea."""

    reason_code = "network_error"
    error_class = "network"
    http_status = None

    def __init__(self, message=None, *, reason_code=None, http_status=None):
        super().__init__(
            message if message is not None else _MSG_NETWORK,
            reason_code=reason_code or "network_error",
            http_status=http_status,
        )


class GiteaConfigError(GiteaClientError):
    """Local configuration / credential resolution failure (not HTTP auth)."""

    reason_code = "config_error"
    error_class = "configuration"
    http_status = None

    def __init__(self, message=None, *, reason_code=None, http_status=None):
        super().__init__(
            message if message is not None else _MSG_CONFIG,
            reason_code=reason_code or "config_error",
            http_status=http_status,
        )


class GiteaHttpError(GiteaClientError):
    """Non-auth HTTP failure with fixed message (no response body)."""

    reason_code = "http_error"
    error_class = "client"
    http_status = None

    def __init__(self, message=None, *, reason_code=None, http_status=None):
        code = reason_code or "http_error"
        if code == "upstream_unavailable":
            fixed = _MSG_UPSTREAM
        else:
            fixed = _MSG_HTTP
            code = "http_error"
        super().__init__(
            message if message is not None else fixed,
            reason_code=code,
            http_status=http_status,
        )


def _looks_like_insufficient_scope(detail: str) -> bool:
    """Internal: inspect redacted body *only* to refine 403 reason_code.

    The body is never stored on the exception or returned to callers.
    """
    lower = (detail or "").lower()
    markers = (
        "insufficient scope",
        "required scope",
        "does not have at least one of required scope",
        "token does not have",
        "missing scope",
        "scope(s)",
    )
    return any(m in lower for m in markers)


def classify_http_status(code: int, *, body_hint: str = "") -> tuple[type, str, int]:
    """Central HTTP status → (exception_class, reason_code, http_status).

    Every HTTP 403 becomes authorization-class. Body text is used only as a
    local hint for scope vs denied reason_code and is never returned.
    """
    if code == 401:
        return (GiteaAuthError, "auth_invalid_token", 401)
    if code == 403:
        if _looks_like_insufficient_scope(body_hint or ""):
            return (GiteaAuthzError, "authz_insufficient_scope", 403)
        return (GiteaAuthzError, "authz_denied", 403)
    if code in (502, 503, 504):
        return (GiteaHttpError, "upstream_unavailable", code)
    return (GiteaHttpError, "http_error", code)


def raise_for_http_status(code: int, body: str = "") -> None:
    """Raise a typed client error for *code* without embedding *body*.

    *body* may be inspected only to choose scope vs denied for 403; it is
    never placed on the exception message.
    """
    # Redact before any inspection; discard after classification.
    try:
        hint = _redact(body or "").strip()
    except Exception:
        hint = ""
    exc_cls, reason, status = classify_http_status(code, body_hint=hint)
    # Explicitly construct without passing body/hint into message.
    if exc_cls is GiteaAuthError:
        raise GiteaAuthError(reason_code=reason, http_status=status)
    if exc_cls is GiteaAuthzError:
        raise GiteaAuthzError(reason_code=reason, http_status=status)
    if reason == "upstream_unavailable":
        raise GiteaHttpError(
            reason_code="upstream_unavailable",
            http_status=status,
        )
    raise GiteaHttpError(reason_code="http_error", http_status=status)


def _raise_http_error(code: int, detail: str = "") -> None:
    """Backward-compatible alias — *detail* is never embedded in the error."""
    raise_for_http_status(code, detail)


def _add_query(url, **params):
    """Return *url* with the given query parameters added or overridden.

    Preserves any existing query string on *url* (e.g. ``?state=open``) so
    pagination params can be layered on top of an already-filtered endpoint.
    """
    parts = urllib.parse.urlsplit(url)
    query = dict(urllib.parse.parse_qsl(parts.query, keep_blank_values=True))
    for key, value in params.items():
        query[str(key)] = str(value)
    new_query = urllib.parse.urlencode(query)
    return urllib.parse.urlunsplit(
        (parts.scheme, parts.netloc, parts.path, new_query, parts.fragment)
    )


def parse_retry_after(value, now=None):
    """Parse a ``Retry-After`` header into a non-negative delay in seconds.

    Supports both forms defined by RFC 7231:
    - a non-negative integer number of seconds (e.g. ``"120"``)
    - an HTTP-date (e.g. ``"Wed, 21 Oct 2015 07:28:00 GMT"``)

    Returns ``None`` when *value* is missing, blank, or unparseable, so the
    caller can fall back to computed backoff. Past dates clamp to ``0``.
    """
    if value is None:
        return None
    value = value.strip()
    if not value:
        return None

    # Seconds form (integer). Reject non-integer numerics like "1.5".
    try:
        seconds = int(value)
        return max(0, seconds)
    except ValueError:
        pass

    # HTTP-date form.
    try:
        when = parsedate_to_datetime(value)
    except (TypeError, ValueError):
        return None
    if when is None:
        return None
    if when.tzinfo is None:
        # RFC dates without a zone are UTC.
        when = when.replace(tzinfo=datetime.timezone.utc)

    now_ts = now if now is not None else time.time()
    return max(0.0, when.timestamp() - now_ts)


def backoff_delay(attempt, base=DEFAULT_BASE_DELAY, cap=DEFAULT_MAX_DELAY, rand=random.random):
    """Full-jitter exponential backoff delay in seconds for a 0-indexed *attempt*.

    Returns a random value in ``[0, min(cap, base * 2**attempt)]``. Full jitter
    spreads retries across the whole window to avoid a thundering herd.
    """
    ceiling = min(cap, base * (2 ** attempt))
    return rand() * ceiling


def api_request(method, url, auth_header, payload=None, *,
                max_retries=None, base_delay=None, max_delay=None,
                timeout=None,
                sleep_func=time.sleep, rand_func=random.random,
                now_func=time.time):
    """Make an authenticated JSON request to the Gitea API.

    Returns parsed JSON on success (or ``None`` for an empty body), and raises
    a classified client error on failure.

    On HTTP 429 the request is retried up to *max_retries* times: honoring a
    valid ``Retry-After`` header (seconds or HTTP-date) when present, otherwise
    using capped jittered exponential backoff. Successful responses are
    unchanged.

    Failures raise typed exceptions with **fixed messages only** (#699). HTTP
    response bodies are read solely for local 403 reason refinement and are
    never stored on exceptions or returned to callers:

    - HTTP 401 → :class:`GiteaAuthError` (``auth_invalid_token``)
    - HTTP 403 → :class:`GiteaAuthzError` (scope or denied)
    - 502/503/504 → :class:`GiteaHttpError` (``upstream_unavailable``)
    - Other non-429 HTTP → :class:`GiteaHttpError` (``http_error``)
    - Timeouts / DNS / ``URLError`` → :class:`GiteaNetworkError`
    - Malformed success JSON → plain ``RuntimeError`` (programming/protocol;
      not reclassified as authentication)

    The ``*_func`` parameters and ``timeout`` are injection points for
    deterministic testing.
    """
    if max_retries is None:
        max_retries = DEFAULT_MAX_RETRIES
    if base_delay is None:
        base_delay = DEFAULT_BASE_DELAY
    if max_delay is None:
        max_delay = DEFAULT_MAX_DELAY
    if timeout is None:
        timeout = DEFAULT_HTTP_TIMEOUT

    data = json.dumps(payload).encode("utf-8") if payload is not None else None
    req = urllib.request.Request(url, data=data, method=method)
    req.add_header("Authorization", auth_header)
    req.add_header("Content-Type", "application/json")
    req.add_header("User-Agent", "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36")

    attempt = 0
    while True:
        try:
            with urllib.request.urlopen(req, timeout=timeout) as resp:
                body = resp.read().decode("utf-8")
        except urllib.error.HTTPError as e:
            if e.code == 429 and attempt < max_retries:
                header = e.headers.get("Retry-After") if e.headers else None
                delay = parse_retry_after(header, now=now_func())
                if delay is None:
                    delay = backoff_delay(attempt, base_delay, max_delay, rand_func)
                attempt += 1
                sleep_func(delay)
                continue
            try:
                error_body = e.read().decode("utf-8", errors="replace")
            except Exception:
                error_body = ""
            # Classify from status (+ local body hint). Body is not embedded.
            try:
                raise_for_http_status(e.code, error_body)
            except GiteaClientError:
                raise
            # Defensive: raise_for_http_status always raises.
            raise GiteaHttpError(http_status=e.code) from e  # pragma: no cover
        except (urllib.error.URLError, TimeoutError) as e:
            # Fixed message only — do not embed URLError reason (may leak paths).
            raise GiteaNetworkError(reason_code="network_error") from e

        if not body:
            return None
        try:
            return json.loads(body)
        except ValueError as e:
            # Programming/protocol failure — not authentication.
            raise RuntimeError("malformed JSON response from Gitea") from e


def api_get_all(url, auth_header, *, limit=None, page_size=50, max_pages=100,
                **kwargs):
    """Fetch a paginated Gitea collection, following page-based pagination.

    Issues successive ``GET`` requests with ``page`` and ``limit`` (per-page)
    query parameters, accumulating list items until one of:

    - a page returns fewer items than the page size (the last page),
    - an empty or ``None`` page is returned (also treated as the end — this is
      how missing/malformed pagination metadata degrades safely),
    - *limit* total items have been collected, or
    - *max_pages* pages have been fetched (a safety cap against runaway loops).

    Pagination relies on the *length of each returned page*, not on
    ``X-Total-Count`` / ``Link`` headers, so it tolerates missing or malformed
    pagination metadata. Returns a list (possibly empty). Raises ``RuntimeError``
    (via :func:`api_request`) on network/HTTP/malformed failures, or if a page is
    not a JSON list. Extra ``kwargs`` pass through to :func:`api_request`.
    """
    if page_size < 1:
        page_size = 1
    if page_size > 50:
        page_size = 50  # Gitea caps per-page results at 50
    if limit is not None and limit < page_size:
        page_size = max(1, limit)

    results = []
    for page in range(1, max_pages + 1):
        page_url = _add_query(url, page=page, limit=page_size)
        data = api_request("GET", page_url, auth_header, **kwargs)
        if data is None:
            break
        if not isinstance(data, list):
            raise RuntimeError(
                f"expected a list page from Gitea, got {type(data).__name__}"
            )
        results.extend(data)
        if limit is not None and len(results) >= limit:
            return results[:limit]
        if len(data) < page_size:
            break
    return results


def api_fetch_page(url, auth_header, *, page=1, limit=50, **kwargs):
    """Fetch one page from a Gitea list endpoint with explicit pagination metadata.

    Returns ``(items, pagination)`` where *pagination* includes ``has_more``,
    ``next_page``, and ``is_final_page`` derived from the returned page length.
    """
    page = max(1, int(page))
    limit = max(1, min(50, int(limit)))
    page_url = _add_query(url, page=page, limit=limit)
    data = api_request("GET", page_url, auth_header, **kwargs)
    if data is None:
        pagination = {
            "page": page,
            "per_page": limit,
            "returned_count": 0,
            "has_more": False,
            "next_page": None,
            "is_final_page": True,
        }
        return [], pagination
    if not isinstance(data, list):
        raise RuntimeError(
            f"expected a list page from Gitea, got {type(data).__name__}"
        )
    returned = len(data)
    has_more = returned >= limit
    pagination = {
        "page": page,
        "per_page": limit,
        "returned_count": returned,
        "has_more": has_more,
        "next_page": page + 1 if has_more else None,
        "is_final_page": not has_more,
    }
    return data, pagination


def gitea_url(host, path):
    """Build a full URL for *host* and *path*, using http for loopback and https for others."""
    if not path.startswith("/"):
        path = "/" + path
    if host.startswith("http://") or host.startswith("https://"):
        return f"{host.rstrip('/')}{path}"
    # Use HTTP for loopback targets, HTTPS for external
    is_loopback = False
    clean_host = host.split(":")[0]
    if clean_host in ("localhost", "127.0.0.1", "::1") or clean_host.startswith("127."):
        is_loopback = True
    scheme = "http" if is_loopback else "https"
    return f"{scheme}://{host}{path}"


def repo_api_url(host, org, repo):
    """Return the base API URL for a repo: https://host/api/v1/repos/org/repo"""
    return gitea_url(host, f"/api/v1/repos/{org}/{repo}")


def get_profile():
    """Return safe runtime *profile* metadata for this MCP process.

    A runtime profile is how the same server code is launched as separate MCP
    entries (e.g. ``gitea-tools-author`` vs ``gitea-tools-reviewer``): each
    process is configured with its own token *and* its own profile name via
    environment variables. This function reads only the non-secret profile
    metadata:

    - ``GITEA_PROFILE_NAME``         — a human label for the running profile.
    - ``GITEA_ALLOWED_OPERATIONS``   — optional comma-separated operation
      categories (descriptive only; not enforced here).
    - ``GITEA_FORBIDDEN_OPERATIONS`` — optional comma-separated operation
      categories this profile must not perform (descriptive only).
    - ``GITEA_AUDIT_LABEL``          — optional short label for audit records.
    - ``GITEA_TOKEN_SOURCE``         — optional *name* of the secret source
      (e.g. an env var name). This is a name only, never a token value.
    - ``GITEA_BASE_URL``             — optional informational base URL.

    It never reads, returns, or logs ``GITEA_TOKEN`` or any credential. The
    token continues to be resolved separately by ``get_auth_header`` and is
    never part of this metadata. Callers may surface the result safely.

    A JSON runtime-profile config (``GITEA_MCP_CONFIG`` + ``GITEA_MCP_PROFILE``,
    see ``gitea_config``) may supply these same fields as a base layer. Explicit
    environment variables always override the JSON profile; the JSON profile
    only fills fields the environment leaves unset. With no config configured,
    behaviour is exactly the environment-only behaviour above.

    Returns:
        dict with 'profile_name', 'allowed_operations' (list),
        'forbidden_operations' (list), 'audit_label', 'token_source_name',
        'base_url', 'username', and 'default_owner'. ``profile_name`` maps to a
        JSON profile's ``execution_profile``; ``token_source_name`` is the
        non-secret auth reference name (env var name or ``keychain:<id>``).
    """
    # JSON layer (base). None when GITEA_MCP_CONFIG is unset; raises ConfigError
    # on a misconfigured file/profile so the problem surfaces clearly at startup.
    jp = gitea_config.resolve_profile() or {}

    def _env_csv(env_key):
        raw = os.environ.get(env_key)
        if raw is None:
            return None
        return [o.strip() for o in raw.split(",") if o.strip()]

    def _json_list(key):
        val = jp.get(key)
        return list(val) if isinstance(val, (list, tuple)) else []

    # profile_name: env > JSON execution_profile > default.
    name = (os.environ.get("GITEA_PROFILE_NAME")
            or jp.get("execution_profile") or "gitea-default")
    name = str(name).strip() or "gitea-default"

    ops = _env_csv("GITEA_ALLOWED_OPERATIONS")
    if ops is None:
        ops = _json_list("allowed_operations")
    forbidden = _env_csv("GITEA_FORBIDDEN_OPERATIONS")
    if forbidden is None:
        forbidden = _json_list("forbidden_operations")

    audit_label = (os.environ.get("GITEA_AUDIT_LABEL") or "").strip() \
        or (jp.get("audit_label") or None)
    # A *name* of the token source (env var name / keychain id), never a value.
    token_source = (os.environ.get("GITEA_TOKEN_SOURCE") or "").strip() \
        or gitea_config.auth_source_name(jp)
    base_url = os.environ.get("GITEA_BASE_URL") or jp.get("base_url") or None
    auth_type = None
    if isinstance(jp.get("auth"), dict):
        auth_type = jp["auth"].get("type")
    elif token_source:
        if token_source.startswith("keychain:"):
            auth_type = "keychain"
        else:
            auth_type = "env"

    return {
        "profile_name": name,
        "allowed_operations": ops,
        "forbidden_operations": forbidden,
        "audit_label": audit_label,
        "token_source_name": token_source,
        "auth_source_type": auth_type,
        "base_url": base_url,
        "username": jp.get("username") or None,
        "default_owner": jp.get("default_owner") or None,
        "profile_path": jp.get("profile_path") or None,
        "environment": jp.get("environment") or None,
        "service": jp.get("service") or None,
        "identity": jp.get("identity") or None,
        "role": jp.get("role") or None,
        "execution_profile": jp.get("execution_profile") or None,
    }