feat: SSO from GTI Gerenciador and the Gerenciador visual identity

Collaborators logged into GTI Gerenciador open GTI Fiscal without a second password. The panel signs a 60 s JWT (Ed25519; the Hub holds only the public key); the Hub checks signature, iss/aud/exp/nbf, maximum lifetime, nonce against a one-time state cookie set when the flow starts at GET /login (login-CSRF protection) and a single-use jti (sso_used_assertions). Collaborators are provisioned in admin_users with auth_source=GTI_GERENCIADOR and no password; a CHECK constraint makes a passwordless local account or a federated account with a password impossible (migration 0021). Sessions last FEDERATED_SESSION_SECONDS (1 h) and are renewed silently through the panel. The local admin stays as break-glass at /login/local, audited. Logout ends only the Hub session. Every SSO success/rejection (stable reason code), logout and local login is audited with the real client IP, never the token. verify_password now fails closed on a missing hash instead of raising. Visual identity: gti-theme.css is a verbatim copy of the panel's design system (tokens, sidebar, light/dark theme) plus its tema.js, sidebar.js and logo; control-panel.css keeps only Hub components, all colours from the panel tokens. Page structure and features unchanged. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-09-23 12:21:26 -03:00
parent 91a2dda969
commit 64d3fc56e2
21 changed files with 1290 additions and 122 deletions
@@ -110,7 +110,8 @@ def test_login_has_centered_card_and_explicit_field_labels():

    assert response.status_code == 200
    assert 'class="login-page"' in response.text
-    assert 'class="login-card"' in response.text
+    # Same card component as the GTI Gerenciador login (gti-theme.css .cartao).
+    assert 'class="cartao login-card"' in response.text
    assert '<label for="login-username">Usuário</label>' in response.text
    assert '<label for="login-password">Senha</label>' in response.text

@@ -0,0 +1,235 @@
+"""SSO from GTI Gerenciador into the Hub (central/app/sso.py, web.py)."""
+import base64
+import dataclasses
+import time
+import uuid
+from datetime import datetime, timedelta, timezone
+from urllib.parse import parse_qs, urlsplit
+
+import jwt
+import pytest
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat
+from fastapi.testclient import TestClient
+from sqlalchemy import select
+
+from central.app import config, sso
+from central.app.human_auth import verify_password
+from central.app.main import create_app
+from central.app.models import AdminUser, AuditEvent, SsoUsedAssertion, WebSession
+from central.tests.test_control_plane import seed, setup_db
+
+START_URL = "https://gtigerenciador.gtigrupo.com.br/sso/fiscal"
+PRIVATE = Ed25519PrivateKey.generate()
+PUBLIC_B64 = base64.urlsafe_b64encode(PRIVATE.public_key().public_bytes(Encoding.Raw, PublicFormat.Raw)).decode().rstrip("=")
+
+
+@pytest.fixture(autouse=True)
+def sso_settings(monkeypatch):
+    monkeypatch.setattr(config, "settings", dataclasses.replace(config.settings, sso_public_key=PUBLIC_B64, sso_start_url=START_URL, web_cookie_secure=True, federated_session_seconds=3600))
+
+
+@pytest.fixture
+def hub():
+    factory = setup_db()
+    seed(factory)  # creates the LOCAL break-glass admin "operator"
+    client = TestClient(create_app(database_probe=lambda: True, session_factory=factory, bootstrap_token=None), base_url="https://testserver")
+    return client, factory
+
+
+def assertion(state, *, sub="funcionario:7", key=PRIVATE, algorithm="EdDSA", aud=sso.AUDIENCE, iss=sso.ISSUER, iat=None, lifetime=60, jti=None, **extra):
+    now = int(iat if iat is not None else time.time())
+    claims = {"iss": iss, "aud": aud, "sub": sub, "iat": now, "nbf": now, "exp": now + lifetime, "jti": jti or str(uuid.uuid4()),
+              "nonce": state, "login": "maria", "name": "Maria Teste", "perfil": "PADRAO", **extra}
+    return jwt.encode(claims, key, algorithm=algorithm)
+
+
+def start(client, next_path="/"):
+    response = client.get("/login", params={"next": next_path}, follow_redirects=False)
+    return response, response.cookies.get(sso.STATE_COOKIE) or client.cookies.get(sso.STATE_COOKIE)
+
+
+def callback(client, token, next_path="/"):
+    return client.post("/auth/sso/callback", data={"assertion": token, "next": next_path}, follow_redirects=False)
+
+
+def audits(factory, action):
+    with factory() as session:
+        return session.scalars(select(AuditEvent).where(AuditEvent.action == action)).all()
+
+
+def test_login_starts_sso_at_the_panel_with_a_browser_bound_state(hub):
+    client, _ = hub
+    response, state = start(client, "/agents")
+    assert response.status_code == 303
+    target = urlsplit(response.headers["location"])
+    assert f"{target.scheme}://{target.netloc}{target.path}" == START_URL
+    query = parse_qs(target.query)
+    assert query["next"] == ["/agents"] and query["state"] == [state] and len(state) >= 16
+    cookie = next(h for h in response.headers.get_list("set-cookie") if h.startswith(sso.STATE_COOKIE))
+    assert "HttpOnly" in cookie and "Secure" in cookie and "SameSite=none" in cookie and "Path=/auth/sso" in cookie
+
+
+def test_active_employee_enters_without_password_and_is_provisioned(hub):
+    client, factory = hub
+    _, state = start(client, "/agents")
+    response = callback(client, assertion(state), "/agents")
+    assert response.status_code == 303 and response.headers["location"] == "/agents"
+    cookies = response.headers.get_list("set-cookie")
+    session_cookie = next(h for h in cookies if h.startswith(config.settings.web_cookie_name))
+    assert "HttpOnly" in session_cookie and "Secure" in session_cookie and "SameSite=lax" in session_cookie
+    csrf_cookie = next(h for h in cookies if h.startswith("gti_csrf"))
+    assert "Secure" in csrf_cookie and "HttpOnly" not in csrf_cookie
+    page = client.get("/agents")
+    assert page.status_code == 200 and "Maria Teste" in page.text and 'name="password"' not in page.text
+    with factory() as session:
+        user = session.scalar(select(AdminUser).where(AdminUser.external_subject == "funcionario:7"))
+        assert user.auth_source == "GTI_GERENCIADOR" and user.password_hash is None and user.external_role == "PADRAO"
+        web_session = session.scalar(select(WebSession).where(WebSession.admin_user_id == user.id))
+        lifetime = web_session.expires_at.replace(tzinfo=timezone.utc) - datetime.now(timezone.utc)
+        assert timedelta(minutes=55) < lifetime <= timedelta(hours=1)
+    success = audits(factory, "SSO_LOGIN_SUCCESS")
+    assert len(success) == 1 and success[0].target_id == "funcionario:7" and "ip" in success[0].context
+
+
+def test_second_login_reuses_the_provisioned_user_and_refreshes_its_name(hub):
+    client, factory = hub
+    _, state = start(client)
+    callback(client, assertion(state))
+    _, state = start(client)
+    callback(client, assertion(state, name="Maria T. Souza", perfil="ADMIN"))
+    with factory() as session:
+        users = session.scalars(select(AdminUser).where(AdminUser.auth_source == "GTI_GERENCIADOR")).all()
+        assert len(users) == 1 and users[0].display_name == "Maria T. Souza" and users[0].external_role == "ADMIN"
+
+
+@pytest.mark.parametrize("builder,reason", [
+    (lambda s: assertion(s, iat=time.time() - 600), "EXPIRED"),
+    (lambda s: assertion(s, key=Ed25519PrivateKey.generate()), "INVALID_SIGNATURE"),
+    (lambda s: assertion(s, aud="outro-sistema"), "WRONG_AUDIENCE"),
+    (lambda s: assertion(s, iss="impostor"), "WRONG_ISSUER"),
+    (lambda s: jwt.encode({"iss": sso.ISSUER, "aud": sso.AUDIENCE, "sub": "funcionario:7", "nonce": s}, "segredo", algorithm="HS256"), "INVALID_ALGORITHM"),
+    (lambda s: assertion(s, lifetime=3600), "LIFETIME_TOO_LONG"),
+    (lambda s: assertion(s, sub="admin"), "INVALID_SUBJECT"),
+    (lambda s: assertion("outro-navegador-0123456789"), "STATE_MISMATCH"),
+    (lambda s: "nao-e-um-jwt", "MALFORMED"),
+])
+def test_invalid_assertions_are_rejected_and_audited(hub, builder, reason):
+    client, factory = hub
+    _, state = start(client)
+    token = builder(state)
+    response = callback(client, token)
+    assert response.status_code == 401 and config.settings.web_cookie_name not in response.cookies
+    rejected = audits(factory, "SSO_LOGIN_REJECTED")
+    assert [event.context["reason"] for event in rejected] == [reason]
+    assert token not in str(rejected[0].context)
+    with factory() as session:
+        assert session.scalar(select(AdminUser).where(AdminUser.auth_source == "GTI_GERENCIADOR")) is None
+
+
+def test_assertion_without_the_state_cookie_is_rejected(hub):
+    client, factory = hub
+    _, state = start(client)
+    client.cookies.delete(sso.STATE_COOKIE)
+    assert callback(client, assertion(state)).status_code == 401
+    assert audits(factory, "SSO_LOGIN_REJECTED")[0].context["reason"] == "STATE_MISMATCH"
+
+
+def test_replayed_assertion_is_rejected(hub):
+    client, factory = hub
+    _, state = start(client)
+    token = assertion(state)
+    assert callback(client, token).status_code == 303
+    client.cookies.clear()
+    client.cookies.set(sso.STATE_COOKIE, state)  # an attacker replaying with the matching state
+    assert callback(client, token).status_code == 401
+    assert audits(factory, "SSO_LOGIN_REJECTED")[0].context["reason"] == "REPLAY"
+    with factory() as session:
+        assert len(session.scalars(select(SsoUsedAssertion)).all()) == 1
+
+
+@pytest.mark.parametrize("malicious", ["//evil.com/x", "https://evil.com", "javascript:alert(1)", "/\\evil.com", "data:text/html,x", "agents"])
+def test_next_never_leaves_the_hub(hub, malicious):
+    client, _ = hub
+    _, state = start(client)
+    response = callback(client, assertion(state), malicious)
+    assert response.status_code == 303 and response.headers["location"] == "/"
+    login = client.get("/login", params={"next": malicious}, follow_redirects=False)
+    assert parse_qs(urlsplit(login.headers["location"]).query)["next"] == ["/"]
+
+
+def test_user_disabled_in_the_hub_cannot_sso(hub):
+    client, factory = hub
+    _, state = start(client)
+    callback(client, assertion(state))
+    with factory() as session:
+        user = session.scalar(select(AdminUser).where(AdminUser.external_subject == "funcionario:7"))
+        user.is_active = False; session.commit()
+    client.cookies.clear()
+    _, state = start(client)
+    assert callback(client, assertion(state)).status_code == 401
+    assert audits(factory, "SSO_LOGIN_REJECTED")[0].context["reason"] == "DISABLED_IN_FISCAL"
+    assert client.get("/", follow_redirects=False).status_code == 303
+
+
+def test_federated_account_has_no_password_login(hub):
+    client, _ = hub
+    _, state = start(client)
+    callback(client, assertion(state))
+    client.cookies.clear()
+    response = client.post("/login", data={"username": "gerenciador:7", "password": "qualquer-coisa-longa"}, follow_redirects=False)
+    assert response.status_code == 401
+    assert verify_password("x", None) is False
+
+
+def test_break_glass_local_admin_still_works_and_is_audited(hub):
+    client, factory = hub
+    assert client.get("/login/local").status_code == 200
+    response = client.post("/login", data={"username": "operator", "password": "a-secure-password"}, follow_redirects=False)
+    assert response.status_code == 303
+    assert client.get("/").status_code == 200
+    assert len(audits(factory, "LOGIN_LOCAL_SUCCESS")) == 1
+    client.cookies.clear()
+    client.post("/login", data={"username": "operator", "password": "errada-mas-longa"})
+    assert len(audits(factory, "LOGIN_LOCAL_FAILURE")) == 1
+
+
+def test_unauthenticated_page_goes_to_sso_and_keeps_the_destination(hub):
+    client, _ = hub
+    response = client.get("/blockers?status=open", follow_redirects=False)
+    assert response.status_code == 303
+    assert parse_qs(urlsplit(response.headers["location"]).query)["next"] == ["/blockers?status=open"]
+
+
+def test_logout_ends_only_the_hub_session_and_csrf_is_enforced(hub):
+    client, factory = hub
+    _, state = start(client)
+    callback(client, assertion(state))
+    assert client.post("/logout", data={}).status_code == 403  # no CSRF token
+    assert client.get("/").status_code == 200
+    response = client.post("/logout", data={"csrf_token": client.cookies.get("gti_csrf")})
+    assert response.status_code == 200 and "Você saiu do GTI Fiscal" in response.text and "gtigerenciador.gtigrupo.com.br" in response.text
+    assert client.get("/", follow_redirects=False).status_code == 303
+    assert len(audits(factory, "LOGOUT")) == 1
+    _, state = start(client)  # coming back through the panel signs in again, no password
+    assert callback(client, assertion(state)).status_code == 303
+
+
+def test_expired_federated_session_sends_back_through_sso(hub):
+    client, factory = hub
+    _, state = start(client)
+    callback(client, assertion(state))
+    with factory() as session:
+        for row in session.scalars(select(WebSession)).all():
+            row.expires_at = datetime.now(timezone.utc) - timedelta(seconds=1)
+        session.commit()
+    response = client.get("/jobs", follow_redirects=False)
+    assert response.status_code == 303 and response.headers["location"].startswith("/login?")
+
+
+def test_without_sso_configuration_the_hub_keeps_the_local_login(hub, monkeypatch):
+    monkeypatch.setattr(config, "settings", dataclasses.replace(config.settings, sso_public_key=""))
+    client, factory = hub
+    assert client.get("/login").status_code == 200
+    assert callback(client, assertion("x" * 20)).status_code == 401
+    assert audits(factory, "SSO_LOGIN_REJECTED")[0].context["reason"] == "SSO_DISABLED"