Audit fixes: 3 sharpest security highs (SSO takeover, token-scope escalation, TLS pin timing)

backend/netsecops/adapters/http_transport.py

@@ -30,6 +30,7 @@
 
 from __future__ import annotations
 
+import asyncio
 import hashlib
 import json
 import ssl
@@ -130,7 +131,16 @@
     # ── lifecycle ───────────────────────────────────────────────────────
 
     async def connect(self) -> None:
-        """Open the client and pin the certificate. Sends no device-facing request."""
+        """Open the client and verify the certificate pin. Sends no device-facing request.
+
+        The pin is checked here, in a TLS handshake that carries no application data,
+        **before** any request. It used to be checked only after a response came back —
+        so the very first request, the authentication exchange carrying the credential
+        (the PAN-OS keygen password in the URL, the Check Point/FortiManager password in
+        the body), reached the peer before the pin was ever compared. An on-path attacker
+        presenting any certificate received the credential, and the mismatch was raised
+        one request too late. Verifying at connect time closes that window.
+        """
         context = ssl.create_default_context()
         if not self.verify_tls:
             # Management interfaces very often carry a self-signed certificate, and a
@@ -141,13 +151,63 @@
             context.check_hostname = False
             context.verify_mode = ssl.CERT_NONE
 
+        await self._pin_certificate(context)
+
         self._client = httpx.AsyncClient(
             base_url=self.base_url,
             verify=context,
             timeout=httpx.Timeout(self.connect_timeout, read=float(self.connect_timeout)),
             follow_redirects=False,
         )
 
+    async def _pin_certificate(self, context: ssl.SSLContext) -> None:
+        """TLS-handshake the peer, verify (or set) the fingerprint pin, send nothing else.
+
+        Uses a raw connection rather than the httpx client so the pin is established with
+        zero application bytes on the wire — the credential must never leave before the
+        peer is authenticated (FR-COL-10).
+        """
+        try:
+            _reader, writer = await asyncio.wait_for(
+                asyncio.open_connection(
+                    self.host, self.port, ssl=context, server_hostname=self.host
+                ),
+                timeout=self.connect_timeout,
+            )
+        except (OSError, ssl.SSLError, asyncio.TimeoutError) as exc:
+            raise DeviceUnreachableError(
+                f"Could not open a TLS connection to {self.host}: {exc}"
+            ) from exc
+
+        try:
+            ssl_object = writer.get_extra_info("ssl_object")
+            # binary_form works under CERT_NONE, where the parsed dict form does not.
+            der = ssl_object.getpeercert(binary_form=True) if ssl_object is not None else None
+        finally:
+            writer.close()
+            try:
+                await writer.wait_closed()
+            except (OSError, ssl.SSLError):
+                # The peer aborting the closed connection tells us nothing about the pin,
+                # which is already read.
+                pass
+
+        if not der:
+            raise DeviceUnreachableError(
+                f"{self.host} presented no TLS certificate, so its identity could not be "
+                "pinned before sending a credential."
+            )
+
+        observed = certificate_fingerprint(der)
+        self.observed_fingerprint = observed
+        if self.known_fingerprint and observed != self.known_fingerprint:
+            raise CertificateChangedError(
+                f"The TLS certificate for {self.host} has changed. Expected "
+                f"{self.known_fingerprint}, got {observed}. This is a planned renewal or "
+                f"an interception; NetSecOps will not collect until the pin is updated "
+                f"deliberately."
+            )
+
     async def disconnect(self) -> None:
         if self._client is not None:
             await self._client.aclose()

backend/netsecops/core/oidc.py

@@ -102,6 +102,10 @@ class VerifiedIdentity:
 
     subject: str
     email: str | None
+    #: Whether the provider asserted `email_verified: true`. An unverified email is one
+    #: the signing-in user set themselves at many providers, so it must not be trusted to
+    #: match or link an existing account. Absent claim → False (default-deny).
+    email_verified: bool
     full_name: str | None
     groups: tuple[str, ...]
     #: Every claim, for the audit record. Read by nothing that makes a decision.
@@ -380,6 +384,7 @@ def verify_id_token(
     return VerifiedIdentity(
         subject=subject,
         email=_claim_str(claims, "email"),
+        email_verified=_claim_bool(claims, "email_verified"),
         full_name=_claim_str(claims, "name") or _claim_str(claims, "preferred_username"),
         groups=_groups(claims, settings.oidc_group_claim),
         claims=claims,
@@ -391,6 +396,21 @@ def _claim_str(claims: dict[str, Any], key: str) -> str | None:
     return value.strip() or None if isinstance(value, str) else None
 
 
+def _claim_bool(claims: dict[str, Any], key: str) -> bool:
+    """A boolean claim, true only when the provider affirmatively asserts it.
+
+    OIDC defines `email_verified` as a boolean, but some providers render it as the
+    string "true"/"false". Anything else — absent, null, a non-affirmative value — is
+    False, so a missing claim never reads as verified.
+    """
+    value = claims.get(key)
+    if isinstance(value, bool):
+        return value
+    if isinstance(value, str):
+        return value.strip().lower() == "true"
+    return False
+
+
 def _groups(claims: dict[str, Any], claim_name: str) -> tuple[str, ...]:
     """The group claim, which providers render three different ways.
 

backend/netsecops/services/auth.py

@@ -26,7 +26,7 @@
     PermissionDeniedError,
 )
 from netsecops.core.logging import get_logger
 from netsecops.core.rbac import GROUP_SCOPED_ROLES, Permission, Principal, Role, Scope
 from netsecops.core.security import (
     TokenType,
     create_access_token,
@@ -709,7 +709,14 @@
         return Principal(
             id=owner.id,
             username=owner.username,
-            roles=owner.role_set or frozenset({Role.API_SERVICE}),
+            # The owner's actual roles — no fallback. `or frozenset({Role.API_SERVICE})`
+            # re-granted the API_SERVICE permission ceiling to an owner whose roles had
+            # been deliberately stripped to disable the token, so the "disabled" token kept
+            # working with escalated rights (2026-09-30 audit). A role-less owner now
+            # yields no permissions (permissions = union of role perms & token_scopes),
+            # so the token authenticates but can do nothing; re-enable it by assigning a
+            # role, disable it by revoking the token.
+            roles=owner.role_set,
             scope=await self.scope_for_user(owner),
             is_service_account=True,
             token_id=stored.id,
@@ -721,7 +728,13 @@
     async def scope_for_user(self, user: User) -> Scope:
         """Resolve object-level visibility for a user (FR-AUTH-05)."""
         roles = user.role_set
-        if not roles or not roles.issubset(GROUP_SCOPED_ROLES):
+        if not roles:
+            # No role grants no visibility. Folding this into the `Scope.all()` branch
+            # below meant a user (or service account) whose roles were all stripped saw
+            # the entire estate — the escalation the 2026-09-30 audit found. A role-less
+            # principal sees nothing.
+            return Scope(unrestricted=False, device_group_ids=frozenset())
+        if not roles.issubset(GROUP_SCOPED_ROLES):
             # Any unrestricted role (Super Admin, Security Analyst) lifts group scoping.
             return Scope.all()
         return Scope(

backend/netsecops/services/sso.py

@@ -328,10 +328,17 @@ async def _resolve_user(
             )
         ).scalar_one_or_none()
 
-        if user is None and identity.email:
+        if user is None and identity.email and identity.email_verified:
             # First sign-in for an account an administrator has already created. Matched
             # on email, case-insensitively, because directories and humans disagree
             # about capitalisation and nothing else in the assertion is stable enough.
+            #
+            # Only when the provider asserted `email_verified: true`. Many providers let a
+            # user set an arbitrary, unverified profile email, so matching (and then
+            # linking the subject to) an existing account on an unverified address is
+            # account takeover: an attacker sets their IdP email to a Super Admin's
+            # address and is handed that account. An unverified email is treated as no
+            # match — the account must be pre-linked by subject or linked by an admin.
             user = (
                 await self.session.execute(
                     select(User).where(func.lower(User.email) == identity.email.strip().lower())

backend/tests/test_http_transport.py

@@ -327,6 +327,47 @@
         with pytest.raises(CertificateChangedError, match="has changed"):
             transport._check_certificate(response)
 
+    async def test_connect_pins_before_building_the_client(self, monkeypatch) -> None:
+        """The pin is verified in a TLS handshake at connect time — before any request —
+        so a changed or attacker certificate is refused before the credential-bearing
+        auth exchange is ever sent (2026-09-30 audit). A matching pin lets connect
+        proceed; a changed one raises before the client (and thus any request) exists.
+        """
+        der = b"the-peer-certificate"
+        fingerprint = certificate_fingerprint(der)
+
+        class _SSLObject:
+            def getpeercert(self, binary_form: bool = False) -> bytes:
+                return der
+
+        class _Writer:
+            def get_extra_info(self, key: str):
+                return _SSLObject() if key == "ssl_object" else None
+
+            def close(self) -> None:
+                pass
+
+            async def wait_closed(self) -> None:
+                return None
+
+        async def fake_open_connection(host, port, *, ssl, server_hostname):  # noqa: ANN001
+            return object(), _Writer()
+
+        monkeypatch.setattr(
+            "netsecops.adapters.http_transport.asyncio.open_connection", fake_open_connection
+        )
+
+        ok = HttpTransport("10.0.0.1", HttpCredentials(), known_fingerprint=fingerprint)
+        await ok.connect()
+        assert ok.observed_fingerprint == fingerprint
+        await ok.disconnect()
+
+        changed = HttpTransport("10.0.0.1", HttpCredentials(), known_fingerprint="SHA256:DE:AD")
+        with pytest.raises(CertificateChangedError, match="has changed"):
+            await changed.connect()
+        # No client was built, so no request — and no credential — could have been sent.
+        assert changed._client is None
+
     def test_the_message_names_both_fingerprints(self) -> None:
         """An operator confirming a planned renewal needs to see the new one, and one
         investigating an interception needs to see both."""

backend/tests/test_sso.py

@@ -78,6 +78,7 @@ def id_token(**overrides: Any) -> str:
         "iat": now,
         "exp": now + 300,
         "email": "dana@example.com",
+        "email_verified": True,
         "name": "Dana Okafor",
         "groups": ["net-admins"],
         "nonce": "NONCE",
@@ -456,6 +457,41 @@ async def test_the_email_match_ignores_capitalisation(self, service, idp, sessio
 
         assert user.external_idp_subject == "idp-subject-001"
 
+    @pytest.mark.anyio
+    async def test_an_unverified_email_does_not_take_over_an_account(
+        self, service, idp, session
+    ) -> None:
+        """A provider that lets a user set an arbitrary, unverified profile email must not
+        let an attacker match — and permanently link their subject to — an existing
+        account by claiming its address. The email fallback requires email_verified."""
+        user = await _seed_dana(session)  # external_idp_subject is None
+        begun, row = await _begin(service)
+        idp.token_response = {
+            "id_token": id_token(nonce=row.nonce, sub="attacker-sub", email_verified=False)
+        }
+
+        with pytest.raises(AuthenticationError):
+            await service.complete(code="c", state=begun.state)
+
+        await session.refresh(user)
+        assert user.external_idp_subject is None  # not linked to the attacker
+
+    @pytest.mark.anyio
+    async def test_a_missing_email_verified_claim_is_treated_as_unverified(
+        self, service, idp, session
+    ) -> None:
+        user = await _seed_dana(session)
+        begun, row = await _begin(service)
+        idp.token_response = {
+            "id_token": id_token(nonce=row.nonce, sub="attacker-sub", email_verified=None)
+        }
+
+        with pytest.raises(AuthenticationError):
+            await service.complete(code="c", state=begun.state)
+
+        await session.refresh(user)
+        assert user.external_idp_subject is None
+
     @pytest.mark.anyio
     async def test_a_second_subject_claiming_a_linked_address_is_refused(
         self, service, idp, session

backend/tests/test_users_service.py

@@ -609,3 +609,48 @@ async def test_password_change_revokes_sessions(
             .all()
         )
         assert rows and all(r.revoked_at is not None for r in rows)
+
+
+class TestApiTokenDoesNotEscalateWhenRolesAreStripped:
+    """Stripping a service account's roles to disable it must neuter its token, not
+    escalate it. The 2026-09-30 audit found a role-less owner re-granted the API_SERVICE
+    ceiling and an UNRESTRICTED device scope."""
+
+    async def test_scope_for_user_with_no_roles_is_empty_not_unrestricted(
+        self, session: AsyncSession
+    ) -> None:
+        from netsecops.services.auth import AuthService
+
+        user = await make_user(session, username="roleless", roles=set())
+        scope = await AuthService(session).scope_for_user(user)
+
+        assert scope.unrestricted is False
+        assert scope.device_group_ids == frozenset()
+
+    async def test_a_role_stripped_token_owner_becomes_powerless(
+        self, service: UserService, session: AsyncSession, super_admin: User
+    ) -> None:
+        from netsecops.services.auth import AuthService
+
+        owner = await make_user(session, username="svc-siem", roles={Role.API_SERVICE})
+        owner.is_service_account = True
+        await session.flush()
+        issued = await service.create_api_token(
+            name="siem-token",
+            owner=owner,
+            scopes={Permission.JOB_EXECUTE, Permission.FINDING_WRITE},
+            expires_at=None,
+            actor=principal(super_admin),
+        )
+        await session.flush()
+
+        # An admin strips every role to disable the account after a suspected leak.
+        await service.set_roles(owner, set(), actor=principal(super_admin))
+        await session.flush()
+
+        resolved = await AuthService(session).principal_for_api_token(issued.plaintext)
+
+        # No API_SERVICE ceiling re-granted, and no unrestricted scope.
+        assert resolved.permissions == frozenset()
+        assert resolved.scope.unrestricted is False
+        assert resolved.scope.device_group_ids == frozenset()