80 lines
2.3 KiB
Python
80 lines
2.3 KiB
Python
from typing import Any
|
|
|
|
from ..npm_client import NpmClient
|
|
from .models import CertificateState
|
|
from .normalizer import (
|
|
normalize_certificate_domains,
|
|
parse_npm_datetime,
|
|
)
|
|
|
|
|
|
class NpmCertificateRepository:
|
|
"""
|
|
Read-only NPM certificate repository.
|
|
|
|
This milestone intentionally discovers metadata only.
|
|
Certificate/private-key material is not requested or logged.
|
|
"""
|
|
|
|
def __init__(self, config):
|
|
self.config = config
|
|
self.client = NpmClient(config)
|
|
|
|
def _rows(self) -> list[dict[str, Any]]:
|
|
data = self.client.certificates()
|
|
|
|
if not isinstance(data, list):
|
|
raise RuntimeError(
|
|
"NPM certificate response was not a list"
|
|
)
|
|
|
|
return data
|
|
|
|
def discover(self) -> list[CertificateState]:
|
|
certificates: list[CertificateState] = []
|
|
|
|
for row in self._rows():
|
|
local_id = row.get("id")
|
|
|
|
if not isinstance(local_id, int):
|
|
raise RuntimeError(
|
|
f"NPM certificate has invalid id: {local_id!r}"
|
|
)
|
|
|
|
meta = row.get("meta")
|
|
if not isinstance(meta, dict):
|
|
meta = {}
|
|
|
|
certificates.append(
|
|
CertificateState(
|
|
identity=normalize_certificate_domains(
|
|
row.get("domain_names")
|
|
),
|
|
local_id=local_id,
|
|
provider=str(
|
|
row.get("provider") or ""
|
|
),
|
|
nice_name=str(
|
|
row.get("nice_name") or ""
|
|
),
|
|
valid_from=parse_npm_datetime(
|
|
row.get("valid_from")
|
|
),
|
|
expires_on=parse_npm_datetime(
|
|
row.get("expires_on")
|
|
),
|
|
cluster_managed=(
|
|
meta.get("cluster_managed") is True
|
|
),
|
|
acme_renew=(
|
|
meta.get("acme_renew") is True
|
|
),
|
|
dns_challenge=(
|
|
meta.get("dns_challenge") is True
|
|
),
|
|
source=self.config.proxy_url,
|
|
)
|
|
)
|
|
|
|
return certificates
|