Add more supported l5d paths for libpvarki

.bumpversion.cfg

@@ -1,5 +1,5 @@
 [bumpversion]
-current_version = 2.2.1
+current_version = 2.2.2
 commit = False
 tag = False
 

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "libpvarki"
-version = "2.2.1"
+version = "2.2.2"
 description = "Common helpers like standard logging init"
 authors = ["Eero af Heurlin <eero.afheurlin@iki.fi>"]
 homepage = "https://github.com/pvarki/python-libpvarki/"

src/libpvarki/__init__.py

@@ -1,3 +1,3 @@
 """Common helpers like standard logging init"""
 
-__version__ = "2.2.1"  # NOTE Use `bump2version --config-file patch` to bump versions correctly
+__version__ = "2.2.2"  # NOTE Use `bump2version --config-file patch` to bump versions correctly

src/libpvarki/middleware/mtlsheader.py

@@ -36,15 +36,10 @@ async def __call__(self, request: Request) -> Optional[DNDict]:  # type: ignore[
         l5d_header = CONFIG("MTLS_L5D_HEADER_NAME", default="l5d-client-id").lower()
         trust_l5d = CONFIG("MTLS_TRUST_L5D", cast=bool, default=False)
 
-        payload: Optional[DNDict] = None
-        if trust_l5d and (l5d_value := request.headers.get(l5d_header)):
-            # Linkerd identity is a bare SPIFFE-style name, not an RFC4514 DN
-            payload = {"CN": l5d_value}
-        elif header_value := request.headers.get(header_name):
-            try:
-                payload = x509name2dict(x509.Name.from_rfc4514_string(header_value))
-            except Exception as exc:
-                raise HTTPException(status_code=403, detail="Invalid authentication") from exc
+        if trust_l5d:
+            payload = self._meshed_payload(request, l5d_header, header_name)
+        else:
+            payload = self._cert_payload(request, header_name)
 
         if payload is None:
             if self.auto_error:
@@ -55,6 +50,36 @@ async def __call__(self, request: Request) -> Optional[DNDict]:  # type: ignore[
         request.state.mtlsdn = payload
         return payload
 
+    def _meshed_payload(self, request: Request, l5d_header: str, header_name: str) -> Optional[DNDict]:
+        """l5d is trusted when present. MTLS_REQUIRE_L5D makes it mandatory (fully-meshed hardening)."""
+        trusted_ingress = {
+            ident.strip() for ident in CONFIG("MTLS_TRUSTED_INGRESS_IDENTITIES", default="").split(",") if ident.strip()
+        }
+        require_l5d = CONFIG("MTLS_REQUIRE_L5D", cast=bool, default=False)
+        l5d_value = request.headers.get(l5d_header)
+        if not l5d_value:
+            # No verified mesh peer. Fail closed only when the mesh is complete (require_l5d);
+            # otherwise honor the cert header so not-yet-meshed callers still authenticate.
+            if require_l5d:
+                return None
+            return self._cert_payload(request, header_name)
+        if l5d_value in trusted_ingress:
+            # Via a trusted ingress: real identity is the forwarded client cert (else None -> JWT).
+            return self._cert_payload(request, header_name)
+        # Lateral in-mesh service call: the peer identity is the client.
+        # Linkerd identity is a bare SPIFFE-style name, not an RFC4514 DN.
+        return {"CN": l5d_value}
+
+    def _cert_payload(self, request: Request, header_name: str) -> Optional[DNDict]:
+        """Parse the proxy-injected client-cert DN header (RFC4514), if present."""
+        header_value = request.headers.get(header_name)
+        if not header_value:
+            return None
+        try:
+            return x509name2dict(x509.Name.from_rfc4514_string(header_value))
+        except Exception as exc:
+            raise HTTPException(status_code=403, detail="Invalid authentication") from exc
+
 
 def x509name2dict(attrs: x509.Name) -> DNDict:
     """Take the Sequence of NameAttributes and make a dict"""

tests/middleware/conftest.py

@@ -0,0 +1,16 @@
+"""Shared fixtures for the middleware tests"""
+
+from typing import Generator
+
+import pytest
+from fastapi.testclient import TestClient
+
+from .app import APP
+
+MTLS_CLIENT_DN = "CN=harjoitus1.pvarki.fi,O=harjoitus1.pvarki.fi,L=KeskiSuomi,ST=Jyvaskyla,C=FI"
+
+
+@pytest.fixture
+def mtlsclient() -> Generator[TestClient, None, None]:
+    """Client presenting a proxy-injected mTLS cert header (legacy / non-l5d path)."""
+    yield TestClient(APP, headers={"X-ClientCert-DN": MTLS_CLIENT_DN})

tests/middleware/test_middleware.py

@@ -1,52 +1,85 @@
 """Test the middleware"""
 
-from typing import Generator
-import logging
+from typing import Dict, Optional
 
 import pytest
 from fastapi.testclient import TestClient
 
 from .app import APP
 
-LOGGER = logging.getLogger(__name__)
+TRUSTED_INGRESS = "traefik.traefik-system.serviceaccount.identity.linkerd.cluster.local"
+LATERAL_SERVICE = "tak.app-tak.serviceaccount.identity.linkerd.cluster.local"
+USER_CERT_DN = "CN=harjoitus1.pvarki.fi,O=harjoitus1.pvarki.fi,L=KeskiSuomi,ST=Jyvaskyla,C=FI"
+USER_CN = "harjoitus1.pvarki.fi"
+SPOOF_DN = "CN=admin,O=admin,C=FI"
 
-# pylint: disable=W0621
-
-
-@pytest.fixture
-def mtlsclient() -> Generator[TestClient, None, None]:
-    """Fake the Nginx header"""
-    client = TestClient(
-        APP,
-        headers={
-            "X-ClientCert-DN": "CN=harjoitus1.pvarki.fi,O=harjoitus1.pvarki.fi,L=KeskiSuomi,ST=Jyvaskyla,C=FI",
-        },
-    )
-    yield client
+# l5d trusted (Traefik registered as an ingress); STRICT additionally requires an l5d header.
+L5D = {"MTLS_TRUST_L5D": "true", "MTLS_TRUSTED_INGRESS_IDENTITIES": TRUSTED_INGRESS}
+L5D_STRICT = {**L5D, "MTLS_REQUIRE_L5D": "true"}
+MTLS_ENV_VARS = ("MTLS_TRUST_L5D", "MTLS_REQUIRE_L5D", "MTLS_TRUSTED_INGRESS_IDENTITIES")
 
 
 def test_hello() -> None:
-    """Check the hello endpoint"""
-    client = TestClient(APP)
-    resp = client.get("/api/v1")
+    """Unauthenticated endpoint is reachable."""
+    resp = TestClient(APP).get("/api/v1")
     assert resp.status_code == 200
-    payload = resp.json()
-    assert payload["message"] == "Hello World"
+    assert resp.json()["message"] == "Hello World"
 
 
-def test_unauth() -> None:
-    """Check that unauth call to auth endpoint fails"""
-    client = TestClient(APP)
-    resp = client.get("/api/v1/check_auth")
-    assert resp.status_code == 403
+@pytest.mark.parametrize(
+    "env, headers, status, cn",
+    [
+        pytest.param({}, {}, 403, None, id="legacy-noauth"),
+        pytest.param({}, {"X-ClientCert-DN": USER_CERT_DN}, 200, USER_CN, id="legacy-cert"),
+        pytest.param(L5D, {"l5d-client-id": LATERAL_SERVICE}, 200, LATERAL_SERVICE, id="lateral-service"),
+        pytest.param(
+            L5D,
+            {"l5d-client-id": TRUSTED_INGRESS, "X-ClientCert-DN": USER_CERT_DN},
+            200,
+            USER_CN,
+            id="ingress-cert",
+        ),
+        pytest.param(L5D, {"l5d-client-id": TRUSTED_INGRESS}, 403, None, id="ingress-nocert"),
+        pytest.param(
+            L5D,
+            {"l5d-client-id": LATERAL_SERVICE, "X-ClientCert-DN": SPOOF_DN},
+            200,
+            LATERAL_SERVICE,
+            id="spoofed-cert-ignored",
+        ),
+        pytest.param(L5D, {"X-ClientCert-DN": USER_CERT_DN}, 200, USER_CN, id="migration-unmeshed-cert"),
+        pytest.param(
+            {"MTLS_TRUST_L5D": "false"},
+            {"l5d-client-id": LATERAL_SERVICE, "X-ClientCert-DN": USER_CERT_DN},
+            200,
+            USER_CN,
+            id="l5d-disabled-uses-cert",
+        ),
+        pytest.param(
+            L5D_STRICT,
+            {"l5d-client-id": TRUSTED_INGRESS, "X-ClientCert-DN": USER_CERT_DN},
+            200,
+            USER_CN,
+            id="strict-ingress-cert",
+        ),
+        pytest.param(L5D_STRICT, {"X-ClientCert-DN": SPOOF_DN}, 403, None, id="strict-no-l5d-fails"),
+    ],
+)
+def test_mtls_identity(
+    monkeypatch: pytest.MonkeyPatch,
+    env: Dict[str, str],
+    headers: Dict[str, str],
+    status: int,
+    cn: Optional[str],
+) -> None:
+    """Resolve the authenticated identity across auth modes: cert header, l5d service, ingress, strict."""
+    for var in MTLS_ENV_VARS:
+        monkeypatch.delenv(var, raising=False)
+    for key, value in env.items():
+        monkeypatch.setenv(key, value)
 
+    resp = TestClient(APP, headers=headers).get("/api/v1/check_auth")
 
-def test_auth(mtlsclient: TestClient) -> None:
-    """Test that the fake header works"""
-    resp = mtlsclient.get("/api/v1/check_auth")
-    assert resp.status_code == 200
-    payload = resp.json()
-    assert payload["ok"]
-    assert "cert" in payload
-    assert "CN" in payload["cert"]
-    assert payload["cert"]["CN"] == "harjoitus1.pvarki.fi"
+    assert resp.status_code == status
+    if cn is not None:
+        assert resp.json()["cert"]["CN"] == cn

tests/middleware/test_product.py

@@ -10,7 +10,6 @@
 
 from libpvarki.schemas.product import UserCRUDRequest, UserInstructionFragment
 from libpvarki.schemas.generic import OperationResultResponse
-from .test_middleware import mtlsclient  # pylint: disable=W0611
 
 
 # pylint: disable=W0621

tests/test_libpvarki.py

@@ -5,4 +5,4 @@
 
 def test_version() -> None:
     """Make sure version matches expected"""
-    assert __version__ == "2.2.1"
+    assert __version__ == "2.2.2"