ankn
a5f8fb49ab
Bahn: aisupport, Analyse-O2C-C2S, awesome-bahn-mcp-servers, beam-mcp,
Confluence_Bot, db-planet-mcp-server, O2C-Harness, project-audit,
Projekt-KIQ-HP, teamlandkarte-mcp
Dhive: Jury-Voting
Privat: CV, NoteGraph (NOTE: NoteGraph needs complete redo after consolidation)
Shared: AI-Orchestrator, OrgMyLife, power_skills_and_more
Shared/references: symphony (read-only)
Bahn repos remain available as independent remotes - this monorepo
pulls them in via subtree, the originals are untouched.
561 lines
20 KiB
Python
561 lines
20 KiB
Python
"""Expiry Checker module for the PAT Manager system.
|
|
|
|
Queries each service API to determine PAT status and classifies tokens
|
|
as healthy, expiring soon, expired, or check failed.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import os
|
|
from datetime import date, timedelta
|
|
|
|
import httpx
|
|
|
|
from .models import CheckResult, PatEntry, PatStatus
|
|
|
|
|
|
def classify_by_date(
|
|
expiry_date: str, reference_date: date, window: int
|
|
) -> tuple[PatStatus, int]:
|
|
"""Pure function that classifies a PAT based on its expiry date.
|
|
|
|
Compares the expiry_date against the reference_date and window to determine
|
|
the token's status.
|
|
|
|
Args:
|
|
expiry_date: ISO 8601 date string (YYYY-MM-DD) representing token expiry.
|
|
reference_date: The date to compare against (typically today).
|
|
window: Number of days before expiry to classify as "expiring soon".
|
|
|
|
Returns:
|
|
A tuple of (status, days_remaining) where:
|
|
- status is one of PatStatus.EXPIRED, EXPIRING_SOON, or HEALTHY
|
|
- days_remaining is max(0, days until expiry)
|
|
|
|
Classification rules:
|
|
- "expired" if expiry_date < reference_date
|
|
- "expiring soon" if reference_date <= expiry_date <= reference_date + timedelta(days=window)
|
|
- "healthy" if expiry_date > reference_date + timedelta(days=window)
|
|
"""
|
|
parsed_expiry = date.fromisoformat(expiry_date)
|
|
days_remaining = (parsed_expiry - reference_date).days
|
|
|
|
if parsed_expiry < reference_date:
|
|
return PatStatus.EXPIRED, 0
|
|
elif parsed_expiry <= reference_date + timedelta(days=window):
|
|
return PatStatus.EXPIRING_SOON, days_remaining
|
|
else:
|
|
return PatStatus.HEALTHY, days_remaining
|
|
|
|
|
|
class ExpiryChecker:
|
|
"""Checks PAT expiry status across all registered services.
|
|
|
|
Attributes:
|
|
expiry_window_days: Number of days before expiry to classify as "expiring soon".
|
|
timeout: HTTP request timeout in seconds.
|
|
max_retries: Maximum number of retry attempts for failed API calls.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
expiry_window_days: int = 14,
|
|
timeout: int = 30,
|
|
max_retries: int = 2,
|
|
) -> None:
|
|
self.expiry_window_days = expiry_window_days
|
|
self.timeout = timeout
|
|
self.max_retries = max_retries
|
|
|
|
async def check_all(
|
|
self, entries: list[PatEntry], secrets: dict[str, str]
|
|
) -> list[CheckResult]:
|
|
"""Check expiry status for all PAT entries.
|
|
|
|
Iterates all entries and calls the appropriate service-specific check
|
|
method for each one. Returns exactly one CheckResult per entry.
|
|
|
|
Args:
|
|
entries: List of PAT entries to check.
|
|
secrets: Mapping of environment variable names to token values
|
|
(e.g. {"GITLAB_PAT": "glpat-xxx", "JIRA_PAT": "token"}).
|
|
|
|
Returns:
|
|
A list of CheckResult objects, one per entry, each with exactly
|
|
one status from {healthy, expiring soon, expired, check failed}.
|
|
"""
|
|
results: list[CheckResult] = []
|
|
|
|
# Map service names to their secret environment variable names
|
|
service_secret_map: dict[str, str] = {
|
|
"GitLab": "GITLAB_PAT",
|
|
"Jira": "JIRA_PAT",
|
|
"Confluence": "CONFLUENCE_PAT",
|
|
"OrgMyLife": "ORGMYLIFE_API_KEY",
|
|
}
|
|
|
|
for entry in entries:
|
|
secret_key = service_secret_map.get(entry.service, "")
|
|
token = secrets.get(secret_key, "")
|
|
|
|
if not token:
|
|
results.append(
|
|
CheckResult(
|
|
entry=entry,
|
|
status=PatStatus.CHECK_FAILED,
|
|
days_remaining=None,
|
|
error_message=f"No secret found for service '{entry.service}' (expected env var: {secret_key})",
|
|
)
|
|
)
|
|
continue
|
|
|
|
try:
|
|
if entry.service == "GitLab":
|
|
result = await self.check_gitlab(entry, token)
|
|
elif entry.service == "Jira":
|
|
result = await self.check_jira(entry, token)
|
|
elif entry.service == "Confluence":
|
|
result = await self.check_confluence(entry, token)
|
|
elif entry.service == "OrgMyLife":
|
|
result = await self.check_orgmylife(entry, token)
|
|
else:
|
|
result = CheckResult(
|
|
entry=entry,
|
|
status=PatStatus.CHECK_FAILED,
|
|
days_remaining=None,
|
|
error_message=f"Unsupported service: {entry.service}",
|
|
)
|
|
results.append(result)
|
|
except Exception as exc:
|
|
results.append(
|
|
CheckResult(
|
|
entry=entry,
|
|
status=PatStatus.CHECK_FAILED,
|
|
days_remaining=None,
|
|
error_message=f"Unexpected error checking {entry.service}/{entry.token_name}: {exc}",
|
|
)
|
|
)
|
|
|
|
return results
|
|
|
|
async def _request_with_retry(
|
|
self,
|
|
client: httpx.AsyncClient,
|
|
method: str,
|
|
url: str,
|
|
headers: dict[str, str],
|
|
service: str,
|
|
token_name: str,
|
|
) -> httpx.Response:
|
|
"""Make an HTTP request with retry logic on timeout/connection errors.
|
|
|
|
Retries up to self.max_retries times with 5s backoff between attempts.
|
|
|
|
Args:
|
|
client: The httpx async client to use.
|
|
method: HTTP method (GET, POST, etc.).
|
|
url: The full URL to request.
|
|
headers: Request headers.
|
|
service: Service name for error messages.
|
|
token_name: Token name for error messages.
|
|
|
|
Returns:
|
|
The HTTP response on success.
|
|
|
|
Raises:
|
|
httpx.TimeoutException: If all retries are exhausted due to timeouts.
|
|
httpx.ConnectError: If all retries are exhausted due to connection errors.
|
|
"""
|
|
last_exception: Exception | None = None
|
|
for attempt in range(self.max_retries + 1):
|
|
try:
|
|
response = await client.request(method, url, headers=headers)
|
|
return response
|
|
except (httpx.TimeoutException, httpx.ConnectError) as exc:
|
|
last_exception = exc
|
|
if attempt < self.max_retries:
|
|
await asyncio.sleep(5)
|
|
raise last_exception # type: ignore[misc]
|
|
|
|
def _check_stale_expiry(
|
|
self, entry: PatEntry, reference_date: date
|
|
) -> bool:
|
|
"""Check if the stored expiry_date has passed (stale).
|
|
|
|
Args:
|
|
entry: The PAT entry to check.
|
|
reference_date: The current date.
|
|
|
|
Returns:
|
|
True if the stored expiry_date has passed, False otherwise.
|
|
"""
|
|
if entry.expiry_date is None:
|
|
return False
|
|
try:
|
|
stored_expiry = date.fromisoformat(entry.expiry_date)
|
|
return stored_expiry < reference_date
|
|
except ValueError:
|
|
return False
|
|
|
|
async def check_gitlab(self, entry: PatEntry, token: str) -> CheckResult:
|
|
"""Check a GitLab PAT's expiry status via the GitLab API.
|
|
|
|
Uses GET /personal_access_tokens/self to retrieve the expires_at field
|
|
and classifies the token based on the expiry window.
|
|
|
|
Args:
|
|
entry: The PAT registry entry to check.
|
|
token: The GitLab personal access token value.
|
|
|
|
Returns:
|
|
CheckResult with the classification status.
|
|
"""
|
|
gitlab_url = os.environ.get("GITLAB_URL", "https://gitlab.com")
|
|
url = f"{gitlab_url}/api/v4/personal_access_tokens/self"
|
|
headers = {"PRIVATE-TOKEN": token}
|
|
|
|
try:
|
|
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
|
response = await self._request_with_retry(
|
|
client, "GET", url, headers, entry.service, entry.token_name
|
|
)
|
|
except (httpx.TimeoutException, httpx.ConnectError) as exc:
|
|
return CheckResult(
|
|
entry=entry,
|
|
status=PatStatus.CHECK_FAILED,
|
|
days_remaining=None,
|
|
error_message=f"Service check failed for {entry.service}/{entry.token_name}: {exc}",
|
|
)
|
|
|
|
if response.status_code == 401:
|
|
return CheckResult(
|
|
entry=entry,
|
|
status=PatStatus.EXPIRED,
|
|
days_remaining=0,
|
|
error_message=None,
|
|
)
|
|
|
|
if response.status_code == 403:
|
|
return CheckResult(
|
|
entry=entry,
|
|
status=PatStatus.CHECK_FAILED,
|
|
days_remaining=None,
|
|
error_message=f"Service check failed for {entry.service}/{entry.token_name}: HTTP {response.status_code}",
|
|
)
|
|
|
|
if not (200 <= response.status_code < 300):
|
|
return CheckResult(
|
|
entry=entry,
|
|
status=PatStatus.CHECK_FAILED,
|
|
days_remaining=None,
|
|
error_message=f"Service check failed for {entry.service}/{entry.token_name}: HTTP {response.status_code}",
|
|
)
|
|
|
|
# Successful response — parse expires_at
|
|
data = response.json()
|
|
expires_at = data.get("expires_at")
|
|
reference_date = date.today()
|
|
|
|
if expires_at is None:
|
|
# No expiry date from API — token has no expiry set
|
|
# Check if stored expiry_date is stale
|
|
if self._check_stale_expiry(entry, reference_date):
|
|
return CheckResult(
|
|
entry=entry,
|
|
status=PatStatus.EXPIRED,
|
|
days_remaining=0,
|
|
error_message=None,
|
|
stale_warning=True,
|
|
)
|
|
return CheckResult(
|
|
entry=entry,
|
|
status=PatStatus.HEALTHY,
|
|
days_remaining=None,
|
|
error_message=None,
|
|
)
|
|
|
|
# Classify based on the API-returned expiry date
|
|
status, days_remaining = classify_by_date(
|
|
expires_at, reference_date, self.expiry_window_days
|
|
)
|
|
|
|
# Check for stale stored expiry_date
|
|
stale_warning = False
|
|
if self._check_stale_expiry(entry, reference_date):
|
|
stale_warning = True
|
|
# If API says token is still valid but stored date has passed,
|
|
# classify as expired with stale warning
|
|
if status == PatStatus.HEALTHY or status == PatStatus.EXPIRING_SOON:
|
|
return CheckResult(
|
|
entry=entry,
|
|
status=PatStatus.EXPIRED,
|
|
days_remaining=0,
|
|
error_message=None,
|
|
stale_warning=True,
|
|
)
|
|
|
|
return CheckResult(
|
|
entry=entry,
|
|
status=status,
|
|
days_remaining=days_remaining,
|
|
error_message=None,
|
|
stale_warning=stale_warning,
|
|
)
|
|
|
|
async def check_jira(self, entry: PatEntry, token: str) -> CheckResult:
|
|
"""Check a Jira PAT's expiry status via an authenticated API call.
|
|
|
|
Interprets HTTP 401 as an expired or revoked token.
|
|
|
|
Args:
|
|
entry: The PAT registry entry to check.
|
|
token: The Jira personal access token value.
|
|
|
|
Returns:
|
|
CheckResult with the classification status.
|
|
"""
|
|
jira_url = os.environ.get("JIRA_URL", "")
|
|
if not jira_url:
|
|
return CheckResult(
|
|
entry=entry,
|
|
status=PatStatus.CHECK_FAILED,
|
|
days_remaining=None,
|
|
error_message=f"Service check failed for {entry.service}/{entry.token_name}: JIRA_URL environment variable not set",
|
|
)
|
|
|
|
url = f"{jira_url}/rest/api/2/myself"
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
|
|
try:
|
|
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
|
response = await self._request_with_retry(
|
|
client, "GET", url, headers, entry.service, entry.token_name
|
|
)
|
|
except (httpx.TimeoutException, httpx.ConnectError) as exc:
|
|
return CheckResult(
|
|
entry=entry,
|
|
status=PatStatus.CHECK_FAILED,
|
|
days_remaining=None,
|
|
error_message=f"Service check failed for {entry.service}/{entry.token_name}: {exc}",
|
|
)
|
|
|
|
reference_date = date.today()
|
|
|
|
if response.status_code == 401:
|
|
# Token is expired/revoked
|
|
return CheckResult(
|
|
entry=entry,
|
|
status=PatStatus.EXPIRED,
|
|
days_remaining=0,
|
|
error_message=None,
|
|
)
|
|
|
|
if not (200 <= response.status_code < 300):
|
|
# Non-auth HTTP error
|
|
return CheckResult(
|
|
entry=entry,
|
|
status=PatStatus.CHECK_FAILED,
|
|
days_remaining=None,
|
|
error_message=f"Service check failed for {entry.service}/{entry.token_name}: HTTP {response.status_code}",
|
|
)
|
|
|
|
# API call succeeded — token is working
|
|
# Check for stale expiry_date
|
|
if self._check_stale_expiry(entry, reference_date):
|
|
return CheckResult(
|
|
entry=entry,
|
|
status=PatStatus.EXPIRED,
|
|
days_remaining=0,
|
|
error_message=None,
|
|
stale_warning=True,
|
|
)
|
|
|
|
# If no expiry_date stored, classify as healthy
|
|
if entry.expiry_date is None:
|
|
return CheckResult(
|
|
entry=entry,
|
|
status=PatStatus.HEALTHY,
|
|
days_remaining=None,
|
|
error_message=None,
|
|
)
|
|
|
|
# Classify based on stored expiry_date
|
|
status, days_remaining = classify_by_date(
|
|
entry.expiry_date, reference_date, self.expiry_window_days
|
|
)
|
|
return CheckResult(
|
|
entry=entry,
|
|
status=status,
|
|
days_remaining=days_remaining,
|
|
error_message=None,
|
|
)
|
|
|
|
async def check_confluence(self, entry: PatEntry, token: str) -> CheckResult:
|
|
"""Check a Confluence PAT's expiry status via an authenticated API call.
|
|
|
|
Interprets HTTP 401 as an expired or revoked token.
|
|
|
|
Args:
|
|
entry: The PAT registry entry to check.
|
|
token: The Confluence personal access token value.
|
|
|
|
Returns:
|
|
CheckResult with the classification status.
|
|
"""
|
|
confluence_url = os.environ.get("CONFLUENCE_URL", "")
|
|
if not confluence_url:
|
|
return CheckResult(
|
|
entry=entry,
|
|
status=PatStatus.CHECK_FAILED,
|
|
days_remaining=None,
|
|
error_message=f"Service check failed for {entry.service}/{entry.token_name}: CONFLUENCE_URL environment variable not set",
|
|
)
|
|
|
|
url = f"{confluence_url}/rest/api/user/current"
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
|
|
try:
|
|
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
|
response = await self._request_with_retry(
|
|
client, "GET", url, headers, entry.service, entry.token_name
|
|
)
|
|
except (httpx.TimeoutException, httpx.ConnectError) as exc:
|
|
return CheckResult(
|
|
entry=entry,
|
|
status=PatStatus.CHECK_FAILED,
|
|
days_remaining=None,
|
|
error_message=f"Service check failed for {entry.service}/{entry.token_name}: {exc}",
|
|
)
|
|
|
|
reference_date = date.today()
|
|
|
|
if response.status_code == 401:
|
|
# Token is expired/revoked
|
|
return CheckResult(
|
|
entry=entry,
|
|
status=PatStatus.EXPIRED,
|
|
days_remaining=0,
|
|
error_message=None,
|
|
)
|
|
|
|
if not (200 <= response.status_code < 300):
|
|
# Non-auth HTTP error
|
|
return CheckResult(
|
|
entry=entry,
|
|
status=PatStatus.CHECK_FAILED,
|
|
days_remaining=None,
|
|
error_message=f"Service check failed for {entry.service}/{entry.token_name}: HTTP {response.status_code}",
|
|
)
|
|
|
|
# API call succeeded — token is working
|
|
# Check for stale expiry_date
|
|
if self._check_stale_expiry(entry, reference_date):
|
|
return CheckResult(
|
|
entry=entry,
|
|
status=PatStatus.EXPIRED,
|
|
days_remaining=0,
|
|
error_message=None,
|
|
stale_warning=True,
|
|
)
|
|
|
|
# If no expiry_date stored, classify as healthy
|
|
if entry.expiry_date is None:
|
|
return CheckResult(
|
|
entry=entry,
|
|
status=PatStatus.HEALTHY,
|
|
days_remaining=None,
|
|
error_message=None,
|
|
)
|
|
|
|
# Classify based on stored expiry_date
|
|
status, days_remaining = classify_by_date(
|
|
entry.expiry_date, reference_date, self.expiry_window_days
|
|
)
|
|
return CheckResult(
|
|
entry=entry,
|
|
status=status,
|
|
days_remaining=days_remaining,
|
|
error_message=None,
|
|
)
|
|
|
|
async def check_orgmylife(self, entry: PatEntry, token: str) -> CheckResult:
|
|
"""Check an OrgMyLife token's expiry status via an authenticated API call.
|
|
|
|
Interprets HTTP 401 or HTTP 403 as an expired or revoked token.
|
|
|
|
Args:
|
|
entry: The PAT registry entry to check.
|
|
token: The OrgMyLife API key value.
|
|
|
|
Returns:
|
|
CheckResult with the classification status.
|
|
"""
|
|
orgmylife_url = os.environ.get("ORGMYLIFE_URL", "https://orgmylife.app")
|
|
url = f"{orgmylife_url}/api/tasks"
|
|
headers = {"X-API-Key": token}
|
|
|
|
try:
|
|
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
|
response = await self._request_with_retry(
|
|
client, "GET", url, headers, entry.service, entry.token_name
|
|
)
|
|
except (httpx.TimeoutException, httpx.ConnectError) as exc:
|
|
return CheckResult(
|
|
entry=entry,
|
|
status=PatStatus.CHECK_FAILED,
|
|
days_remaining=None,
|
|
error_message=f"Service check failed for {entry.service}/{entry.token_name}: {exc}",
|
|
)
|
|
|
|
reference_date = date.today()
|
|
|
|
if response.status_code in (401, 403):
|
|
# Token is expired/revoked
|
|
return CheckResult(
|
|
entry=entry,
|
|
status=PatStatus.EXPIRED,
|
|
days_remaining=0,
|
|
error_message=None,
|
|
)
|
|
|
|
if not (200 <= response.status_code < 300):
|
|
# Non-auth HTTP error
|
|
return CheckResult(
|
|
entry=entry,
|
|
status=PatStatus.CHECK_FAILED,
|
|
days_remaining=None,
|
|
error_message=f"Service check failed for {entry.service}/{entry.token_name}: HTTP {response.status_code}",
|
|
)
|
|
|
|
# API call succeeded — token is working
|
|
# Check for stale expiry_date
|
|
if self._check_stale_expiry(entry, reference_date):
|
|
return CheckResult(
|
|
entry=entry,
|
|
status=PatStatus.EXPIRED,
|
|
days_remaining=0,
|
|
error_message=None,
|
|
stale_warning=True,
|
|
)
|
|
|
|
# If no expiry_date stored, classify as healthy
|
|
if entry.expiry_date is None:
|
|
return CheckResult(
|
|
entry=entry,
|
|
status=PatStatus.HEALTHY,
|
|
days_remaining=None,
|
|
error_message=None,
|
|
)
|
|
|
|
# Classify based on stored expiry_date
|
|
status, days_remaining = classify_by_date(
|
|
entry.expiry_date, reference_date, self.expiry_window_days
|
|
)
|
|
return CheckResult(
|
|
entry=entry,
|
|
status=status,
|
|
days_remaining=days_remaining,
|
|
error_message=None,
|
|
)
|