Le champ ntfy_server du formulaire /settings était utilisé tel quel dans
urllib.request.urlopen sans aucune validation. Un attaquant pouvait pointer
le serveur vers 169.254.169.254 (metadata AWS/GCP), le réseau interne, ou un
service protégé, et comme la reminder_loop relance l'URL toutes les 5 min le
SSRF devenait persistant. Le header Authorization étant aussi contrôlable,
des requêtes authentifiées vers des services internes étaient possibles.
Ajout de notifications.is_safe_ntfy_server() qui valide : schéma http(s),
absence de credentials embarqués, refus de localhost et de toutes les IP
littérales privées (loopback, link-local, 10/8, 172.16/12, 192.168/16,
fc00::/7, ::1, multicast, réservé). send_ntfy et POST /settings appellent
ce validateur avant toute ouverture de connexion ou persistance ; un serveur
refusé déclenche une redirection /settings?ssrf_error=1 côté UI.
Les tests simulant un échec réseau (127.0.0.1:1) passent sur un host public
injoignable (.invalid), et un nouveau bloc test_security couvre les cas
loopback/private/bad-scheme/credentials et la non-ouverture de socket vers
une IP privée.
💘 Generated with Crush
Assisted-by: Crush:glm-5.2
165 lines
6.0 KiB
Python
165 lines
6.0 KiB
Python
"""Tests de sécurité : anti-forgery du cookie de session.
|
|
|
|
Avant le correctif, n'importe qui pouvait forger le cookie `user_id=admin` et
|
|
accéder aux données d'un autre utilisateur. La signature HMAC doit rendre toute
|
|
altération détectable.
|
|
"""
|
|
import sys
|
|
|
|
# DATA_DIR temporaire est positionné par conftest.py avant l'import de models.
|
|
import models # noqa: F401
|
|
|
|
sys.path.insert(0, ".")
|
|
import main
|
|
from fastapi.testclient import TestClient
|
|
|
|
client = TestClient(main.app)
|
|
|
|
|
|
def _seed_user(uid: str):
|
|
import bcrypt
|
|
models.save_auth(uid, email=f"{uid}@x.fr",
|
|
password_hash=bcrypt.hashpw(b"x" * 12, bcrypt.gensalt()).decode())
|
|
|
|
|
|
def _login(uid: str) -> dict:
|
|
_seed_user(uid)
|
|
r = client.post("/login/password",
|
|
data={"email": f"{uid}@x.fr", "password": "x" * 12},
|
|
follow_redirects=False)
|
|
assert r.status_code == 302, r.text
|
|
return {"user_id": r.cookies.get("user_id")}
|
|
|
|
|
|
# ── Forgery du cookie ─────────────────────────────────────────────────────────
|
|
|
|
def test_unsigned_cookie_rejected():
|
|
"""Un cookie `user_id=admin` forgé à la main doit être rejeté."""
|
|
_seed_user("alice")
|
|
r = client.get("/semaine/2026/29", cookies={"user_id": "alice"}, follow_redirects=False)
|
|
assert r.status_code == 302
|
|
assert "/login" in r.headers["location"]
|
|
|
|
|
|
def test_tampered_signature_rejected():
|
|
"""Un cookie valide dont on modifie la signature doit être rejeté."""
|
|
cookies = _login("alice")
|
|
raw = cookies["user_id"]
|
|
user_id, _, sig = raw.rpartition(".")
|
|
# Inverse un caractère de la signature
|
|
bad_sig = ("0" if sig[0] != "0" else "1") + sig[1:]
|
|
tampered = f"{user_id}.{bad_sig}"
|
|
r = client.get("/semaine/2026/29", cookies={"user_id": tampered}, follow_redirects=False)
|
|
assert r.status_code == 302
|
|
assert "/login" in r.headers["location"]
|
|
|
|
|
|
def test_valid_signed_cookie_accepted():
|
|
"""Le cookie signé délivré par /login/password doit ouvrir la session."""
|
|
cookies = _login("alice")
|
|
r = client.get("/semaine/2026/29", cookies=cookies, follow_redirects=False)
|
|
assert r.status_code == 200
|
|
|
|
|
|
def test_signed_cookie_cannot_be_replayed_for_other_user():
|
|
"""La signature d'alice ne doit pas valider pour bob (user_id modifié)."""
|
|
cookies = _login("alice")
|
|
raw = cookies["user_id"]
|
|
_, _, sig = raw.rpartition(".")
|
|
# Remplace alice par bob en conservant la signature d'alice
|
|
forged = f"bob.{sig}"
|
|
_seed_user("bob")
|
|
r = client.get("/semaine/2026/29", cookies={"user_id": forged}, follow_redirects=False)
|
|
assert r.status_code == 302
|
|
assert "/login" in r.headers["location"]
|
|
|
|
|
|
# ── Helpers de signature ──────────────────────────────────────────────────────
|
|
|
|
def test_sign_and_verify_roundtrip():
|
|
sig = main._sign_user_id("alice")
|
|
assert main._verify_signed_cookie(sig) == "alice"
|
|
|
|
|
|
def test_verify_rejects_malformed_inputs():
|
|
for bad in ("", "alice", "alice.", ".", "alice.deadbeef", "inva/lid.xxx"):
|
|
assert main._verify_signed_cookie(bad) is None
|
|
|
|
|
|
def test_secret_is_stable_across_calls():
|
|
s1 = models.load_or_create_secret()
|
|
s2 = models.load_or_create_secret()
|
|
assert s1 == s2
|
|
|
|
|
|
# ── Validation SSRF du serveur ntfy ──────────────────────────────────────────
|
|
|
|
import notifications # noqa: E402
|
|
|
|
|
|
def test_ntfy_server_validation_accepts_https():
|
|
assert notifications.is_safe_ntfy_server("https://ntfy.sh")
|
|
assert notifications.is_safe_ntfy_server("https://ntfy.example.com:8080")
|
|
|
|
|
|
def test_ntfy_server_validation_rejects_loopback():
|
|
for bad in ("http://127.0.0.1:1", "http://127.0.0.1", "http://localhost",
|
|
"https://localhost:8443", "https://[::1]"):
|
|
assert not notifications.is_safe_ntfy_server(bad), bad
|
|
|
|
|
|
def test_ntfy_server_validation_rejects_private_networks():
|
|
for bad in ("https://10.0.0.1", "https://172.16.5.5", "https://192.168.1.1",
|
|
"https://169.254.169.254", "http://10.255.255.1:8080"):
|
|
assert not notifications.is_safe_ntfy_server(bad), bad
|
|
|
|
|
|
def test_ntfy_server_validation_rejects_bad_schemes():
|
|
for bad in ("ftp://ntfy.sh", "file:///etc/passwd", "gopher://x", "javascript:alert(1)"):
|
|
assert not notifications.is_safe_ntfy_server(bad), bad
|
|
|
|
|
|
def test_ntfy_server_validation_rejects_credentials_in_url():
|
|
assert not notifications.is_safe_ntfy_server("https://user:pass@ntfy.sh")
|
|
|
|
|
|
def test_ntfy_server_validation_rejects_empty_or_none():
|
|
for bad in ("", None, " "):
|
|
assert not notifications.is_safe_ntfy_server(bad), bad
|
|
|
|
|
|
def test_send_ntfy_refuses_ssrf_target():
|
|
"""Un appel vers une IP privée ne doit jamais ouvrir de connexion réseau."""
|
|
import urllib.request as u
|
|
called = {"n": 0}
|
|
orig = u.urlopen
|
|
try:
|
|
u.urlopen = lambda *a, **kw: called.__setitem__("n", called["n"] + 1)
|
|
ok = notifications.send_ntfy(
|
|
"http://169.254.169.254", "ssrf-topic",
|
|
"msg", "titre", user_id=None, token="",
|
|
)
|
|
finally:
|
|
u.urlopen = orig
|
|
assert ok is False
|
|
assert called["n"] == 0 # urlopen n'a jamais été appelé
|
|
|
|
|
|
def test_settings_save_rejects_ssrf_ntfy_server():
|
|
"""POST /settings avec un serveur SSRF doit être rejeté."""
|
|
cookies = _login("carol")
|
|
# Préserve un topic valide
|
|
models.save_notif_config("carol", ntfy_topic="t", ntfy_server="https://ntfy.sh")
|
|
r = client.post("/settings", cookies=cookies, data={
|
|
"ntfy_topic": "topic-x",
|
|
"ntfy_server": "http://169.254.169.254",
|
|
"ntfy_token": "",
|
|
"rappel_debut_h": "8",
|
|
"rappel_fin_h": "19",
|
|
}, follow_redirects=False)
|
|
assert r.status_code == 303
|
|
assert "ssrf_error=1" in r.headers["location"]
|
|
# La valeur précédente est conservée
|
|
cfg = models.load_notif_config("carol")
|
|
assert cfg["ntfy_server"] == "https://ntfy.sh"
|