feat: bind the GTI Fiscal SSO assertion to the browser (state/nonce)

/sso/fiscal now requires the state that GTI Fiscal creates when the flow
starts and signs it as the "nonce" claim; GTI Fiscal checks it against
its own one-time cookie. Without a valid state the panel sends the
browser to GTI Fiscal's /login, so the flow always starts there. This
prevents someone from pushing an assertion for their own account into
another person's browser (login CSRF).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Gustavo
2026-09-23 12:21:27 -03:00
co-authored by Claude Opus 5.5
parent 9139a1f6f3
commit 161da90e02
3 changed files with 50 additions and 20 deletions
+10 -5
View File
@@ -34,29 +34,34 @@ def _pagina_erro(mensagem: str, status_code: int) -> HTMLResponse:
@router.get("/sso/fiscal") @router.get("/sso/fiscal")
def iniciar_sso_fiscal(request: Request, next: str | None = None): def iniciar_sso_fiscal(request: Request, next: str | None = None, state: str | None = None):
destino = sso.proximo_seguro(next) destino = sso.proximo_seguro(next)
if not sso.configurado(): if not sso.configurado():
return _pagina_erro("O acesso ao GTI Fiscal não está configurado neste painel.", 503) return _pagina_erro("O acesso ao GTI Fiscal não está configurado neste painel.", 503)
# Sem state o fluxo não começou no Fiscal (ex.: clique na sidebar): manda
# para o login do Fiscal, que cria o state e devolve para cá.
if not sso.estado_valido(state):
return RedirectResponse(sso.url_login_fiscal(destino), status_code=303)
parametros = urlencode({"state": state, "next": destino})
if request.url.scheme != "https": if request.url.scheme != "https":
publica = os.environ.get("SSO_GERENCIADOR_URL_PUBLICA", "").rstrip("/") publica = os.environ.get("SSO_GERENCIADOR_URL_PUBLICA", "").rstrip("/")
if publica.startswith("https://"): if publica.startswith("https://"):
return RedirectResponse(f"{publica}/sso/fiscal?{urlencode({'next': destino})}", status_code=303) return RedirectResponse(f"{publica}/sso/fiscal?{parametros}", status_code=303)
return _pagina_erro("O acesso ao GTI Fiscal só funciona pelo endereço HTTPS do painel.", 400) return _pagina_erro("O acesso ao GTI Fiscal só funciona pelo endereço HTTPS do painel.", 400)
try: try:
usuario = exigir_login(request) usuario = exigir_login(request)
except NaoAutenticado: except NaoAutenticado:
volta = f"/sso/fiscal?{urlencode({'next': destino})}" return RedirectResponse(f"/login?{urlencode({'next': f'/sso/fiscal?{parametros}'})}", status_code=303)
return RedirectResponse(f"/login?{urlencode({'next': volta})}", status_code=303)
funcionario = db_funcionarios.obter_por_login(usuario.login) funcionario = db_funcionarios.obter_por_login(usuario.login)
if funcionario is None: if funcionario is None:
db_log.registrar(str(usuario), "sso_fiscal", "funcionario", usuario.login, False, "conta de bootstrap não usa SSO") db_log.registrar(str(usuario), "sso_fiscal", "funcionario", usuario.login, False, "conta de bootstrap não usa SSO")
return _pagina_erro("A conta de bootstrap não acessa o GTI Fiscal — entre com seu usuário de funcionário.", 403) return _pagina_erro("A conta de bootstrap não acessa o GTI Fiscal — entre com seu usuário de funcionário.", 403)
assercao = sso.emitir_assercao(funcionario) assercao = sso.emitir_assercao(funcionario, nonce=state)
db_log.registrar(str(usuario), "sso_fiscal", "funcionario", str(funcionario["ID"]), True) db_log.registrar(str(usuario), "sso_fiscal", "funcionario", str(funcionario["ID"]), True)
resposta = templates.TemplateResponse( resposta = templates.TemplateResponse(
request, "sso_fiscal.html", {"acao": sso.url_callback(), "assercao": assercao, "proximo": destino} request, "sso_fiscal.html", {"acao": sso.url_callback(), "assercao": assercao, "proximo": destino}
+17 -3
View File
@@ -7,7 +7,8 @@ funcionário e entrega ao Fiscal uma asserção assinada, de vida curta.
tem a chave pública, então não consegue emitir asserções. tem a chave pública, então não consegue emitir asserções.
- Claims: iss/aud fixos, sub = "funcionario:<ID>" (o ID de - Claims: iss/aud fixos, sub = "funcionario:<ID>" (o ID de
FUNCIONARIO_PAINEL — o LOGIN é editável, então não serve de identificador), FUNCIONARIO_PAINEL — o LOGIN é editável, então não serve de identificador),
iat/nbf/exp (60 s) e jti único, que o Fiscal grava para recusar replay. iat/nbf/exp (60 s), jti único (o Fiscal grava para recusar replay) e nonce
= state do Fiscal (amarra a asserção ao navegador que iniciou o fluxo).
- Nenhuma senha, hash ou cookie deste painel sai daqui: só nome, login e - Nenhuma senha, hash ou cookie deste painel sai daqui: só nome, login e
perfil, que o Fiscal usa para exibir e para autorização futura. perfil, que o Fiscal usa para exibir e para autorização futura.
- A asserção vai no corpo de um POST auto-submetido (routers/sso.py), nunca - A asserção vai no corpo de um POST auto-submetido (routers/sso.py), nunca
@@ -18,9 +19,10 @@ nem nenhuma outra credencial.
""" """
import base64 import base64
import os import os
import re
import time import time
import uuid import uuid
from urllib.parse import urlsplit from urllib.parse import urlencode, urlsplit
import jwt import jwt
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
@@ -50,6 +52,17 @@ def origem_callback() -> str:
return f"{partes.scheme}://{partes.netloc}" return f"{partes.scheme}://{partes.netloc}"
def estado_valido(valor: str | None) -> bool:
"""state gerado pelo Fiscal (cookie do navegador que iniciou o fluxo); volta
assinado como "nonce", e o Fiscal confere contra o cookie — assim ninguém
consegue empurrar a própria asserção para o navegador de outra pessoa."""
return bool(valor) and bool(re.fullmatch(r"[A-Za-z0-9_\-]{16,128}", valor))
def url_login_fiscal(proximo: str) -> str:
return f"{origem_callback()}/login?{urlencode({'next': proximo})}"
def proximo_seguro(valor: str | None) -> str: def proximo_seguro(valor: str | None) -> str:
"""Só caminho relativo dentro do Fiscal. Qualquer outra coisa vira "/".""" """Só caminho relativo dentro do Fiscal. Qualquer outra coisa vira "/"."""
if ( if (
@@ -70,7 +83,7 @@ def _chave_privada() -> Ed25519PrivateKey:
return Ed25519PrivateKey.from_private_bytes(semente) return Ed25519PrivateKey.from_private_bytes(semente)
def emitir_assercao(funcionario: dict, agora: float | None = None) -> str: def emitir_assercao(funcionario: dict, nonce: str, agora: float | None = None) -> str:
momento = int(agora if agora is not None else time.time()) momento = int(agora if agora is not None else time.time())
claims = { claims = {
"iss": EMISSOR, "iss": EMISSOR,
@@ -80,6 +93,7 @@ def emitir_assercao(funcionario: dict, agora: float | None = None) -> str:
"nbf": momento, "nbf": momento,
"exp": momento + VALIDADE_SEGUNDOS, "exp": momento + VALIDADE_SEGUNDOS,
"jti": str(uuid.uuid4()), "jti": str(uuid.uuid4()),
"nonce": nonce,
"login": funcionario["LOGIN"], "login": funcionario["LOGIN"],
"name": funcionario["NOME"], "name": funcionario["NOME"],
"perfil": funcionario["PERFIL"], "perfil": funcionario["PERFIL"],
+23 -12
View File
@@ -14,6 +14,7 @@ from app.db import log as db_log
from app.main import app from app.main import app
CALLBACK = "https://gtifiscal.gtigrupo.com.br/auth/sso/callback" CALLBACK = "https://gtifiscal.gtigrupo.com.br/auth/sso/callback"
STATE = "estadoDoNavegador_0123456789"
ATIVO = {"ID": 7, "NOME": "Maria Teste", "LOGIN": "maria", "SENHA_HASH": "x", "PERFIL": "PADRAO", "ATIVO": 1} ATIVO = {"ID": 7, "NOME": "Maria Teste", "LOGIN": "maria", "SENHA_HASH": "x", "PERFIL": "PADRAO", "ATIVO": 1}
@@ -42,10 +43,11 @@ def _assercao(resposta):
def test_funcionario_ativo_recebe_formulario_com_assercao_valida(chave, monkeypatch): def test_funcionario_ativo_recebe_formulario_com_assercao_valida(chave, monkeypatch):
resposta = _cliente(monkeypatch=monkeypatch).get("/sso/fiscal?next=/agents") resposta = _cliente(monkeypatch=monkeypatch).get("/sso/fiscal", params={"next": "/agents", "state": STATE})
assert resposta.status_code == 200 assert resposta.status_code == 200
assert f'action="{CALLBACK}"' in resposta.text and 'name="next" value="/agents"' in resposta.text assert f'action="{CALLBACK}"' in resposta.text and 'name="next" value="/agents"' in resposta.text
claims = jwt.decode(_assercao(resposta), chave, algorithms=["EdDSA"], audience="gti-fiscal", issuer="gti-gerenciador", options={"require": ["exp", "iat", "nbf", "iss", "aud", "sub", "jti"]}) claims = jwt.decode(_assercao(resposta), chave, algorithms=["EdDSA"], audience="gti-fiscal", issuer="gti-gerenciador", options={"require": ["exp", "iat", "nbf", "iss", "aud", "sub", "jti", "nonce"]})
assert claims["nonce"] == STATE
assert claims["sub"] == "funcionario:7" and claims["login"] == "maria" and claims["perfil"] == "PADRAO" assert claims["sub"] == "funcionario:7" and claims["login"] == "maria" and claims["perfil"] == "PADRAO"
assert claims["exp"] - claims["iat"] == 60 assert claims["exp"] - claims["iat"] == 60
assert "password" not in claims and "SENHA_HASH" not in str(claims) assert "password" not in claims and "SENHA_HASH" not in str(claims)
@@ -58,55 +60,64 @@ def test_funcionario_ativo_recebe_formulario_com_assercao_valida(chave, monkeypa
def test_cada_emissao_tem_jti_diferente(chave, monkeypatch): def test_cada_emissao_tem_jti_diferente(chave, monkeypatch):
cliente = _cliente(monkeypatch=monkeypatch) cliente = _cliente(monkeypatch=monkeypatch)
a, b = (jwt.decode(_assercao(cliente.get("/sso/fiscal")), chave, algorithms=["EdDSA"], audience="gti-fiscal") for _ in range(2)) a, b = (jwt.decode(_assercao(cliente.get("/sso/fiscal", params={"state": STATE})), chave, algorithms=["EdDSA"], audience="gti-fiscal") for _ in range(2))
assert a["jti"] != b["jti"] assert a["jti"] != b["jti"]
def test_sem_sessao_vai_para_login_e_volta_ao_sso(chave, monkeypatch): def test_sem_sessao_vai_para_login_e_volta_ao_sso(chave, monkeypatch):
resposta = _cliente(logado=False, monkeypatch=monkeypatch).get("/sso/fiscal?next=/jobs", follow_redirects=False) resposta = _cliente(logado=False, monkeypatch=monkeypatch).get("/sso/fiscal", params={"next": "/jobs", "state": STATE}, follow_redirects=False)
assert resposta.status_code == 303 assert resposta.status_code == 303
destino = urlsplit(resposta.headers["location"]) destino = urlsplit(resposta.headers["location"])
assert destino.path == "/login" assert destino.path == "/login"
assert parse_qs(destino.query)["next"] == ["/sso/fiscal?next=%2Fjobs"] assert parse_qs(destino.query)["next"] == [f"/sso/fiscal?state={STATE}&next=%2Fjobs"]
def test_funcionario_inativo_nao_recebe_assercao(chave, monkeypatch): def test_funcionario_inativo_nao_recebe_assercao(chave, monkeypatch):
resposta = _cliente(funcionario={**ATIVO, "ATIVO": 0}, monkeypatch=monkeypatch).get("/sso/fiscal", follow_redirects=False) resposta = _cliente(funcionario={**ATIVO, "ATIVO": 0}, monkeypatch=monkeypatch).get("/sso/fiscal", params={"state": STATE}, follow_redirects=False)
assert resposta.status_code == 303 and resposta.headers["location"].startswith("/login") assert resposta.status_code == 303 and resposta.headers["location"].startswith("/login")
assert "assertion" not in resposta.text assert "assertion" not in resposta.text
def test_conta_de_bootstrap_nao_usa_sso(chave, monkeypatch): def test_conta_de_bootstrap_nao_usa_sso(chave, monkeypatch):
monkeypatch.setenv("ADMIN_USER", "admin") monkeypatch.setenv("ADMIN_USER", "admin")
resposta = _cliente(funcionario=None, monkeypatch=monkeypatch).get("/sso/fiscal", follow_redirects=False) resposta = _cliente(funcionario=None, monkeypatch=monkeypatch).get("/sso/fiscal", params={"state": STATE}, follow_redirects=False)
assert resposta.status_code == 403 and "assertion" not in resposta.text assert resposta.status_code == 403 and "assertion" not in resposta.text
def test_http_redireciona_para_endereco_https_publico(chave, monkeypatch): def test_http_redireciona_para_endereco_https_publico(chave, monkeypatch):
resposta = _cliente(https=False, monkeypatch=monkeypatch).get("/sso/fiscal?next=/blockers", follow_redirects=False) resposta = _cliente(https=False, monkeypatch=monkeypatch).get("/sso/fiscal", params={"next": "/blockers", "state": STATE}, follow_redirects=False)
assert resposta.status_code == 303 assert resposta.status_code == 303
assert resposta.headers["location"] == "https://gtigerenciador.gtigrupo.com.br/sso/fiscal?next=%2Fblockers" assert resposta.headers["location"] == f"https://gtigerenciador.gtigrupo.com.br/sso/fiscal?state={STATE}&next=%2Fblockers"
@pytest.mark.parametrize("malicioso", ["//evil.com/x", "https://evil.com", "javascript:alert(1)", "/\\evil.com", "data:text/html,x", "agents", "/a\nb"]) @pytest.mark.parametrize("malicioso", ["//evil.com/x", "https://evil.com", "javascript:alert(1)", "/\\evil.com", "data:text/html,x", "agents", "/a\nb"])
def test_next_malicioso_vira_raiz(chave, monkeypatch, malicioso): def test_next_malicioso_vira_raiz(chave, monkeypatch, malicioso):
resposta = _cliente(monkeypatch=monkeypatch).get("/sso/fiscal", params={"next": malicioso}) resposta = _cliente(monkeypatch=monkeypatch).get("/sso/fiscal", params={"next": malicioso, "state": STATE})
assert 'name="next" value="/"' in resposta.text assert 'name="next" value="/"' in resposta.text
def test_callback_so_https_configurado(chave, monkeypatch): def test_callback_so_https_configurado(chave, monkeypatch):
monkeypatch.setenv("SSO_FISCAL_CALLBACK_URL", "http://gtifiscal.gtigrupo.com.br/auth/sso/callback") monkeypatch.setenv("SSO_FISCAL_CALLBACK_URL", "http://gtifiscal.gtigrupo.com.br/auth/sso/callback")
with pytest.raises(RuntimeError): with pytest.raises(RuntimeError):
_cliente(monkeypatch=monkeypatch).get("/sso/fiscal") _cliente(monkeypatch=monkeypatch).get("/sso/fiscal", params={"state": STATE})
def test_sem_configuracao_nao_emite(monkeypatch): def test_sem_configuracao_nao_emite(monkeypatch):
monkeypatch.delenv("SSO_FISCAL_CHAVE_PRIVADA", raising=False) monkeypatch.delenv("SSO_FISCAL_CHAVE_PRIVADA", raising=False)
monkeypatch.setattr(db_log, "registrar", lambda *a, **k: None) monkeypatch.setattr(db_log, "registrar", lambda *a, **k: None)
resposta = _cliente(monkeypatch=monkeypatch).get("/sso/fiscal") resposta = _cliente(monkeypatch=monkeypatch).get("/sso/fiscal", params={"state": STATE})
assert resposta.status_code == 503 assert resposta.status_code == 503
def test_outras_paginas_mantem_csp_padrao(chave, monkeypatch): def test_outras_paginas_mantem_csp_padrao(chave, monkeypatch):
resposta = _cliente(monkeypatch=monkeypatch).get("/login") resposta = _cliente(monkeypatch=monkeypatch).get("/login")
assert "form-action 'self'" in resposta.headers["content-security-policy"] assert "form-action 'self'" in resposta.headers["content-security-policy"]
@pytest.mark.parametrize("estado", [None, "curto", "com espaço e símbolos!!!!!!", "x" * 200])
def test_sem_state_valido_manda_para_o_login_do_fiscal(chave, monkeypatch, estado):
params = {"next": "/agents"} | ({"state": estado} if estado is not None else {})
resposta = _cliente(monkeypatch=monkeypatch).get("/sso/fiscal", params=params, follow_redirects=False)
assert resposta.status_code == 303
assert resposta.headers["location"] == "https://gtifiscal.gtigrupo.com.br/login?next=%2Fagents"
assert "assertion" not in resposta.text