From b31beba11f710127052261fc7f68b482ed4628c1 Mon Sep 17 00:00:00 2001 From: Krishnendu De Date: Wed, 30 Sep 2026 06:59:19 +0530 Subject: [PATCH 1/3] Require a verified email before SSO matches an existing account (audit HIGH) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit SSO account-matching fell back to email when no account matched the IdP subject, and on a match with a not-yet-linked account it bound the asserting subject to that account and issued its session — all without checking `email_verified`. Many providers (consumer Google, multi-tenant Entra, self-service directories) let a user set an arbitrary, unverified profile email, so an attacker could set their IdP email to a Super Admin's address and be handed that account: full takeover. - VerifiedIdentity now carries `email_verified`, parsed from the claim (bool or the string "true"; absent/anything else -> False, default-deny). - The email fallback in _resolve_user runs only when the provider asserted email_verified: true. An unverified or absent-verification email is treated as no match — the account must be pre-linked by subject or linked by an administrator. Adds tests: an unverified email (and a missing email_verified claim) with a foreign subject is refused and leaves the target account's external_idp_subject unlinked. The test IdP's default claims now assert email_verified: true, as a correctly-configured provider does. 58 SSO tests pass. Co-Authored-By: Claude Opus 4.8 --- backend/netsecops/core/oidc.py | 20 +++++++++++++++++ backend/netsecops/services/sso.py | 9 +++++++- backend/tests/test_sso.py | 36 +++++++++++++++++++++++++++++++ 3 files changed, 64 insertions(+), 1 deletion(-) diff --git a/backend/netsecops/core/oidc.py b/backend/netsecops/core/oidc.py index f8f1169..a6e93b0 100644 --- a/backend/netsecops/core/oidc.py +++ b/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. diff --git a/backend/netsecops/services/sso.py b/backend/netsecops/services/sso.py index f0d0da2..75e54e7 100644 --- a/backend/netsecops/services/sso.py +++ b/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()) diff --git a/backend/tests/test_sso.py b/backend/tests/test_sso.py index ce8d0eb..ca700f6 100644 --- a/backend/tests/test_sso.py +++ b/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 From b84a3442b3a89512e2f3fe128931ee44eaa963b4 Mon Sep 17 00:00:00 2001 From: Krishnendu De Date: Wed, 30 Sep 2026 07:18:29 +0530 Subject: [PATCH 2/3] Stop a role-stripped service account escalating its API token (audit HIGH) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Stripping every role from a service account — the intuitive way to disable it after a suspected leak — silently ESCALATED its token instead of neutering it, two ways: - principal_for_api_token used `roles=owner.role_set or frozenset({Role.API_SERVICE})`, re-granting the API_SERVICE permission ceiling to an owner whose roles were removed. - scope_for_user folded "no roles" into the `Scope.all()` branch, so a role-less owner got UNRESTRICTED device-group visibility. Together, a "disabled" token kept authenticating with API_SERVICE-tier permissions across EVERY device group. Now: the token carries the owner's actual roles (no fallback), so a role-less owner has no permissions (permissions = union of role perms & token_scopes = {}), and scope_for_user returns an empty, non-unrestricted scope for no roles. The token authenticates but can do nothing; re-enable by assigning a role, disable by revoking the token. Regular users with no roles likewise now see nothing rather than everything. Adds tests: scope_for_user with no roles is empty-not-unrestricted, and a role-stripped token owner resolves to zero permissions and an empty scope. Full auth/rbac/users/scope suite green (1149). Co-Authored-By: Claude Opus 4.8 --- backend/netsecops/services/auth.py | 17 +++++++++-- backend/tests/test_users_service.py | 45 +++++++++++++++++++++++++++++ 2 files changed, 60 insertions(+), 2 deletions(-) diff --git a/backend/netsecops/services/auth.py b/backend/netsecops/services/auth.py index 8a9a13b..b6baa0e 100644 --- a/backend/netsecops/services/auth.py +++ b/backend/netsecops/services/auth.py @@ -709,7 +709,14 @@ async def principal_for_api_token(self, plaintext: str) -> Principal: 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 principal_for_api_token(self, plaintext: str) -> Principal: 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( diff --git a/backend/tests/test_users_service.py b/backend/tests/test_users_service.py index c895f0d..c26b0ce 100644 --- a/backend/tests/test_users_service.py +++ b/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() From 6f77c38e41a6e638c4bbd8f4541e27307e739e70 Mon Sep 17 00:00:00 2001 From: Krishnendu De Date: Wed, 30 Sep 2026 07:22:15 +0530 Subject: [PATCH 3/3] Verify the TLS certificate pin before sending any credential (audit HIGH) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit For a device with verify_tls off (the default for self-signed management interfaces), connect() built an httpx client with CERT_NONE and did no pinning; the fingerprint was compared only in _check_certificate(), called AFTER the request returned. So the very first request — the authentication exchange carrying the credential (PAN-OS keygen password in the URL, Check Point/FortiManager password in the body) — reached the peer before the pin was ever checked. An on-path attacker presenting any certificate received the credential, and the mismatch was raised one request too late. connect() now performs the pin check in a raw TLS handshake (asyncio.open_connection) that carries no application data, before the httpx client is built. A changed/foreign fingerprint raises CertificateChangedError at connect time — no client, no request, no credential on the wire. First-use still pins (observed_fingerprint). The post-response _check_certificate stays as defence-in-depth for a mid-session change. Adds a test: a matching pin lets connect proceed and records the fingerprint; a changed pin raises before the client exists (so no credential-bearing request was possible). 27 http-transport tests pass. Co-Authored-By: Claude Opus 4.8 --- backend/netsecops/adapters/http_transport.py | 62 +++++++++++++++++++- backend/tests/test_http_transport.py | 41 +++++++++++++ 2 files changed, 102 insertions(+), 1 deletion(-) diff --git a/backend/netsecops/adapters/http_transport.py b/backend/netsecops/adapters/http_transport.py index cb167a3..17d4116 100644 --- a/backend/netsecops/adapters/http_transport.py +++ b/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 @@ def base_url(self) -> str: # ── 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,6 +151,8 @@ async def connect(self) -> None: 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, @@ -148,6 +160,54 @@ async def connect(self) -> None: 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() diff --git a/backend/tests/test_http_transport.py b/backend/tests/test_http_transport.py index e5b8a38..4ee822e 100644 --- a/backend/tests/test_http_transport.py +++ b/backend/tests/test_http_transport.py @@ -327,6 +327,47 @@ def getpeercert(self, binary_form: bool = False) -> bytes: 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."""