135 lines
4.9 KiB
Python
135 lines
4.9 KiB
Python
"""Resolve the approved Jira project binding for an identity.
|
|
|
|
The target resolver answers: "which Jira project key is this identity allowed
|
|
to sync/search?" The answer comes from configuration, never from the caller's
|
|
``project_id`` argument. This is the structural guarantee that prevents a
|
|
caller-controlled value from redirecting queries to another project's data.
|
|
|
|
Configuration sources (checked in order):
|
|
1. ``JIRA_KB_PROJECT_MAP`` environment variable (JSON dict mapping
|
|
``org_unit/customer/project`` or bare ``project`` → Jira project key).
|
|
2. ``jira_knowledge.projects`` section in the Cowork config file.
|
|
3. Fallback: the identity's ``project`` field used as-is when it looks like a
|
|
valid Jira project key (uppercase letters/digits with a hyphen).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import re
|
|
from dataclasses import dataclass
|
|
from typing import Optional
|
|
|
|
from ...mcp_servers.project_context.foundation import IdentityContext, ProviderError
|
|
|
|
_JIRA_KEY_PATTERN = re.compile(r"^[A-Z][A-Z0-9]+$")
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class JiraTarget:
|
|
"""Resolved Jira project binding for one identity."""
|
|
jira_project_key: str
|
|
jira_base_url: str
|
|
cowork_project_id: str
|
|
|
|
|
|
class JiraTargetResolver:
|
|
"""Identity → approved Jira project binding.
|
|
|
|
Never trusts caller-supplied routing. If no binding exists for the
|
|
identity, raises ``UNAVAILABLE`` so the provider layer can return a clean
|
|
error instead of silently falling back to the wrong project.
|
|
"""
|
|
|
|
def resolve(self, identity: IdentityContext) -> JiraTarget:
|
|
"""Resolve the Jira target for ``identity``.
|
|
|
|
Raises:
|
|
ProviderError: When no binding is configured or the identity's
|
|
project is not mapped to an approved Jira project.
|
|
"""
|
|
base_url = self._resolve_base_url()
|
|
if not base_url:
|
|
raise ProviderError(
|
|
"UNAVAILABLE",
|
|
"Jira base URL is not configured for this environment.",
|
|
retryable=False,
|
|
)
|
|
|
|
project_key = self._resolve_project_key(identity)
|
|
if not project_key:
|
|
raise ProviderError(
|
|
"UNAVAILABLE",
|
|
f"No Jira project binding is configured for identity '{identity.project}'.",
|
|
retryable=False,
|
|
)
|
|
|
|
return JiraTarget(
|
|
jira_project_key=project_key,
|
|
jira_base_url=base_url,
|
|
cowork_project_id=identity.project,
|
|
)
|
|
|
|
def _resolve_base_url(self) -> str:
|
|
"""Jira base URL from env or config."""
|
|
env = os.environ.get("JIRA_KB_BASE_URL", "").strip().rstrip("/")
|
|
if env:
|
|
return env
|
|
try:
|
|
from ...infrastructure.config.json_config_repository import JsonConfigRepository
|
|
cfg = JsonConfigRepository()
|
|
jira_cfg = cfg.data.get("jira", {}) or {}
|
|
url = str(jira_cfg.get("base_url", "") or "").strip().rstrip("/")
|
|
return url
|
|
except Exception: # noqa: BLE001
|
|
return ""
|
|
|
|
def _resolve_project_key(self, identity: IdentityContext) -> str:
|
|
"""Map the identity to its approved Jira project key."""
|
|
# 1. Environment variable map (for CI / container deployments).
|
|
env_map = self._load_env_map()
|
|
identity_key = f"{identity.org_unit}/{identity.customer}/{identity.project}"
|
|
key = env_map.get(identity_key) or env_map.get(identity.project)
|
|
if key and _JIRA_KEY_PATTERN.match(key):
|
|
return key
|
|
|
|
# 2. Config file map.
|
|
cfg_map = self._load_config_map()
|
|
key = cfg_map.get(identity_key) or cfg_map.get(identity.project)
|
|
if key and _JIRA_KEY_PATTERN.match(key):
|
|
return key
|
|
|
|
# 3. Fallback: identity.project itself if it looks like a Jira key.
|
|
if _JIRA_KEY_PATTERN.match(identity.project):
|
|
return identity.project
|
|
|
|
return ""
|
|
|
|
@staticmethod
|
|
def _load_env_map() -> dict[str, str]:
|
|
raw = os.environ.get("JIRA_KB_PROJECT_MAP", "").strip()
|
|
if not raw:
|
|
return {}
|
|
try:
|
|
parsed = json.loads(raw)
|
|
except json.JSONDecodeError:
|
|
return {}
|
|
if not isinstance(parsed, dict):
|
|
return {}
|
|
return {str(k): str(v) for k, v in parsed.items() if isinstance(k, str) and isinstance(v, str)}
|
|
|
|
@staticmethod
|
|
def _load_config_map() -> dict[str, str]:
|
|
try:
|
|
from ...infrastructure.config.json_config_repository import JsonConfigRepository
|
|
cfg = JsonConfigRepository()
|
|
jk = cfg.data.get("jira_knowledge", {}) or {}
|
|
projects = jk.get("projects", {}) or {}
|
|
if isinstance(projects, dict):
|
|
return {str(k): str(v) for k, v in projects.items()}
|
|
except Exception: # noqa: BLE001
|
|
pass
|
|
return {}
|
|
|
|
|
|
__all__ = ["JiraTarget", "JiraTargetResolver"] |