86408887e5
Meniu hamburger reorganizat pe grupuri cu etichete (LUCRU / CONT) cu iconite SVG discrete; "Cont" redenumit "Setari" (URL ?tab=cont pastrat, zero rute moarte). Pagina Setari restructurata in sub-taburi interne: Date firma / Credentiale RAR / Cheie API / Cont (plan + schimbare parola), cu indicator de stare per tab. Sub-tabul activ e pastrat server-side (activ_subtab) dupa fiecare swap HTMX; comutarea intre taburi e client-side, delegata pe document (supravietuieste swap-urilor). Credentiale RAR: mediile Testare/Productie ca doua carduri paralele. Schimbare parola de acces din sub-tabul Cont (functie noua, lipsea): - users.change_password (scrypt): verifica parola actuala constant-time, valideaza noua (10..128, diferita), rescrie hash+salt nou; - ruta POST /cont/schimba-parola, scoped pe user_id din sesiune, CSRF enforce, parola niciodata re-pusa in raspuns, jurnal 'parola_schimbata'. Referinte copy "Meniu cont > Cont" -> "Setari". Teste: schimbare parola (succes / parola actuala gresita / confirmare nepotrivita / lungime) + izolare rate-limit login in fixture-ul clientului. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
400 lines
15 KiB
Python
400 lines
15 KiB
Python
"""Sectiunea 'Contul meu' — rotire cheie API + creds RAR din UI.
|
|
|
|
Rute testate:
|
|
- GET /_fragments/cont -> card "Contul meu"
|
|
- POST /cont/roteste-cheie -> cheie noua afisata o singura data
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import re
|
|
import tempfile
|
|
|
|
import pytest
|
|
from starlette.testclient import TestClient
|
|
|
|
|
|
@pytest.fixture()
|
|
def client(monkeypatch):
|
|
"""Client fara web_auth_required (dev mode) — sesiunea se seteaza manual."""
|
|
tmp = tempfile.mkdtemp()
|
|
monkeypatch.setenv("AUTOPASS_DB_PATH", os.path.join(tmp, "t.db"))
|
|
from app.config import get_settings
|
|
get_settings.cache_clear()
|
|
# Izolare rate-limit login intre teste (limiter global per-IP pe viata procesului):
|
|
# fara reset, seria de _login din fisier depaseste login_rate_max si da 429 (fals negativ).
|
|
from app.web import ratelimit
|
|
ratelimit._hits.clear()
|
|
from app.main import app
|
|
with TestClient(app, follow_redirects=False) as c:
|
|
yield c
|
|
get_settings.cache_clear()
|
|
|
|
|
|
@pytest.fixture()
|
|
def client_prod(monkeypatch):
|
|
"""Client cu web_auth_required=True (mod prod) — CSRF enforce."""
|
|
tmp = tempfile.mkdtemp()
|
|
monkeypatch.setenv("AUTOPASS_DB_PATH", os.path.join(tmp, "t.db"))
|
|
monkeypatch.setenv("AUTOPASS_WEB_AUTH_REQUIRED", "true")
|
|
from app.config import get_settings
|
|
get_settings.cache_clear()
|
|
from app.main import app
|
|
with TestClient(app, follow_redirects=False) as c:
|
|
yield c
|
|
get_settings.cache_clear()
|
|
|
|
|
|
def _create_account_user(email: str = "user@test.com", password: str = "parolasecreta10"):
|
|
"""Creeaza cont + user + cheie API initiala. Intoarce (acct_id, user_id, api_key)."""
|
|
from app.accounts import create_account
|
|
from app.users import create_user
|
|
from app.auth import create_api_key
|
|
from app.db import get_connection
|
|
|
|
conn = get_connection()
|
|
try:
|
|
acct_id = create_account(conn, "Service Test", active=True)
|
|
user_id = create_user(conn, acct_id, email, password)
|
|
api_key = create_api_key(conn, acct_id)
|
|
return acct_id, user_id, api_key
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def _login(client, email: str, password: str) -> None:
|
|
"""Face login real prin HTTP si seteaza cookie-ul de sesiune pe client."""
|
|
# Obtine CSRF token de pe pagina de login
|
|
resp = client.get("/login")
|
|
assert resp.status_code == 200
|
|
m = re.search(r'name="csrf_token"\s+value="([^"]+)"', resp.text)
|
|
if not m:
|
|
m = re.search(r'value="([^"]+)"\s+name="csrf_token"', resp.text)
|
|
assert m, "csrf_token negasit pe /login"
|
|
csrf = m.group(1)
|
|
|
|
resp = client.post("/login", data={
|
|
"email": email,
|
|
"parola": password,
|
|
"csrf_token": csrf,
|
|
})
|
|
# 303 redirect la / inseamna login reusit
|
|
assert resp.status_code == 303, f"Login esuat: {resp.status_code} {resp.text[:200]}"
|
|
|
|
|
|
def _get_csrf_from_fragment(client) -> str:
|
|
"""Obtine CSRF token din fragmentul /_fragments/cont."""
|
|
resp = client.get("/_fragments/cont")
|
|
assert resp.status_code == 200
|
|
m = re.search(r'name="csrf_token"\s+value="([^"]+)"', resp.text)
|
|
if not m:
|
|
m = re.search(r'value="([^"]+)"\s+name="csrf_token"', resp.text)
|
|
assert m, f"csrf_token negasit in /_fragments/cont: {resp.text[:500]}"
|
|
return m.group(1)
|
|
|
|
|
|
# ============================================================
|
|
# test_roteste_cheie_afisata_o_data
|
|
# ============================================================
|
|
|
|
def test_roteste_cheie_afisata_o_data(client):
|
|
"""User logat roteste cheia: raspunsul contine 'rfak_'; cheia veche revocata."""
|
|
acct_id, user_id, api_key_initiala = _create_account_user("roteste@test.com")
|
|
_login(client, "roteste@test.com", "parolasecreta10")
|
|
|
|
csrf = _get_csrf_from_fragment(client)
|
|
resp = client.post("/cont/roteste-cheie", data={"csrf_token": csrf})
|
|
|
|
assert resp.status_code == 200
|
|
assert "rfak_" in resp.text, f"Cheia noua nu apare in raspuns: {resp.text[:500]}"
|
|
|
|
# Verifica in DB: cheia veche revocata, una noua activa
|
|
from app.db import get_connection
|
|
conn = get_connection()
|
|
try:
|
|
rows = conn.execute(
|
|
"SELECT id, active FROM api_keys WHERE account_id=? ORDER BY id",
|
|
(acct_id,),
|
|
).fetchall()
|
|
# Trebuie sa avem minim 2 chei: cea initiala (active=0) si cea noua (active=1)
|
|
active_keys = [r for r in rows if r["active"] == 1]
|
|
inactive_keys = [r for r in rows if r["active"] == 0]
|
|
assert len(active_keys) == 1, f"Trebuia exact 1 cheie activa, gasit: {len(active_keys)}"
|
|
assert len(inactive_keys) >= 1, "Cheia veche trebuia revocata (active=0)"
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
# ============================================================
|
|
# test_roteste_fara_csrf_403_in_prod
|
|
# ============================================================
|
|
|
|
def test_roteste_fara_csrf_403_in_prod(client_prod):
|
|
"""Prod + sesiune autentificata + CSRF lipsa -> 403."""
|
|
# Cream cont + user
|
|
acct_id, user_id, _ = _create_account_user("csrf_test@test.com")
|
|
|
|
# Login real
|
|
_login(client_prod, "csrf_test@test.com", "parolasecreta10")
|
|
|
|
# POST fara csrf_token (sau cu token gresit)
|
|
resp = client_prod.post("/cont/roteste-cheie", data={"csrf_token": "token_gresit"})
|
|
assert resp.status_code == 403, f"Trebuia 403, got {resp.status_code}"
|
|
|
|
|
|
# ============================================================
|
|
# test_fragment_cont_nelogat_redirect
|
|
# ============================================================
|
|
|
|
def test_fragment_cont_nelogat_redirect(monkeypatch):
|
|
"""Fara sesiune + web_auth_required=True -> 303 redirect /login."""
|
|
tmp = tempfile.mkdtemp()
|
|
monkeypatch.setenv("AUTOPASS_DB_PATH", os.path.join(tmp, "t_nl.db"))
|
|
monkeypatch.setenv("AUTOPASS_WEB_AUTH_REQUIRED", "true")
|
|
from app.config import get_settings
|
|
get_settings.cache_clear()
|
|
from app.main import app
|
|
with TestClient(app, follow_redirects=False) as c:
|
|
resp = c.get("/_fragments/cont")
|
|
assert resp.status_code == 303
|
|
assert "/login" in resp.headers.get("location", "")
|
|
get_settings.cache_clear()
|
|
|
|
|
|
# ============================================================
|
|
# Sectiunea 'Date firma' + banner cont incomplet
|
|
# ============================================================
|
|
|
|
def _create_complete_account(
|
|
name: str = "Firma Test SRL",
|
|
login_email: str = "firma_test@test.com",
|
|
account_email: str = "contact@firma.com",
|
|
cui: str = "RO12345678",
|
|
password: str = "parolasecreta10",
|
|
):
|
|
"""Creeaza cont COMPLET (name+email+CUI) + user. Intoarce (acct_id, user_id)."""
|
|
from app.accounts import create_account
|
|
from app.users import create_user
|
|
from app.db import get_connection
|
|
|
|
conn = get_connection()
|
|
try:
|
|
acct_id = create_account(conn, name, cui=cui, email=account_email, active=True)
|
|
user_id = create_user(conn, acct_id, login_email, password)
|
|
return acct_id, user_id
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def test_cont_afiseaza_companie_email_cui(client):
|
|
"""Fragment /_fragments/cont contine sectiunea 'Date firma' cu companie, email, CUI prefilled."""
|
|
_create_complete_account(
|
|
name="Test Firma SRL",
|
|
login_email="tfirma@test.com",
|
|
account_email="contact_tf@test.com",
|
|
cui="RO11111111",
|
|
)
|
|
_login(client, "tfirma@test.com", "parolasecreta10")
|
|
|
|
resp = client.get("/_fragments/cont")
|
|
assert resp.status_code == 200
|
|
assert "Date firma" in resp.text or "date-firma" in resp.text, \
|
|
f"Sectiunea 'Date firma' lipseste: {resp.text[:500]}"
|
|
assert "Test Firma SRL" in resp.text, f"Compania nu e prefilled: {resp.text[:500]}"
|
|
assert "contact_tf@test.com" in resp.text, f"Email-ul nu e prefilled: {resp.text[:500]}"
|
|
assert "RO11111111" in resp.text, f"CUI-ul nu e prefilled: {resp.text[:500]}"
|
|
|
|
|
|
def test_post_date_firma_actualizeaza(client):
|
|
"""POST /cont/date-firma actualizeaza accounts.name, accounts.email, accounts.cui in DB."""
|
|
acct_id, user_id, _ = _create_account_user("update_df@test.com")
|
|
_login(client, "update_df@test.com", "parolasecreta10")
|
|
|
|
csrf = _get_csrf_from_fragment(client)
|
|
|
|
resp = client.post("/cont/date-firma", data={
|
|
"csrf_token": csrf,
|
|
"companie": "Firma Actualizata SRL",
|
|
"email": "contact@firma-act.com",
|
|
"cui": "RO99887766",
|
|
})
|
|
assert resp.status_code == 200, f"POST /cont/date-firma a returnat {resp.status_code}"
|
|
|
|
# Verifica in DB ca datele au fost actualizate
|
|
from app.db import get_connection
|
|
conn = get_connection()
|
|
try:
|
|
row = conn.execute(
|
|
"SELECT name, email, cui FROM accounts WHERE id=?", (acct_id,)
|
|
).fetchone()
|
|
assert row["name"] == "Firma Actualizata SRL", f"name neschimbat: {row['name']}"
|
|
assert row["email"] == "contact@firma-act.com", f"email neschimbat: {row['email']}"
|
|
assert row["cui"] == "RO99887766", f"cui neschimbat: {row['cui']}"
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def test_post_date_firma_cui_duplicat_eroare(client):
|
|
"""POST /cont/date-firma cu CUI deja folosit de alt cont -> eroare in raspuns."""
|
|
# Cont A cu CUI existent
|
|
_create_complete_account(
|
|
name="Firma A SRL",
|
|
login_email="firma_a_dup@test.com",
|
|
account_email="a_dup@test.com",
|
|
cui="ROAAA11111",
|
|
)
|
|
# Cont B fara CUI
|
|
acct_b, user_b, _ = _create_account_user("firma_b_dup@test.com")
|
|
|
|
_login(client, "firma_b_dup@test.com", "parolasecreta10")
|
|
csrf = _get_csrf_from_fragment(client)
|
|
|
|
resp = client.post("/cont/date-firma", data={
|
|
"csrf_token": csrf,
|
|
"companie": "Firma B SRL",
|
|
"email": "firma_b_dup@test.com",
|
|
"cui": "ROAAA11111", # CUI-ul lui A — duplicat
|
|
})
|
|
assert resp.status_code == 200
|
|
text = resp.text.lower()
|
|
assert "deja" in text or "duplicat" in text or "folosit" in text or "eroare" in text, \
|
|
f"Mesaj eroare CUI duplicat lipsa: {resp.text[:500]}"
|
|
|
|
# Contul B nu trebuie sa aiba CUI-ul lui A in DB
|
|
from app.db import get_connection
|
|
conn = get_connection()
|
|
try:
|
|
row = conn.execute("SELECT cui FROM accounts WHERE id=?", (acct_b,)).fetchone()
|
|
assert row["cui"] != "ROAAA11111", "CUI-ul duplicat a fost totusi salvat in DB"
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def test_banner_cont_incomplet_pe_legacy(client):
|
|
"""Acasa afiseaza banner 'Completeaza datele firmei' cand contul e incomplet (fara email/CUI)."""
|
|
# Cont fara email/CUI (legacy: creat fara aceste campuri)
|
|
_create_account_user("legacy_test@test.com")
|
|
_login(client, "legacy_test@test.com", "parolasecreta10")
|
|
|
|
resp = client.get("/")
|
|
assert resp.status_code == 200
|
|
text = resp.text.lower()
|
|
# Banner trebuie sa apara cand contul e incomplet
|
|
assert "completeaza" in text or "date firm" in text or "incomplet" in text, \
|
|
f"Banner 'Completeaza datele firmei' lipsa pe Acasa: {resp.text[:2000]}"
|
|
|
|
|
|
# ============================================================
|
|
# Sub-tab Cont: schimbare parola de acces (PRD 5.21)
|
|
# ============================================================
|
|
|
|
def _get_csrf_from_cont(client) -> str:
|
|
"""CSRF token din panoul Setari (tab=cont)."""
|
|
resp = client.get("/?tab=cont")
|
|
assert resp.status_code == 200
|
|
m = re.search(r'name="csrf_token"\s+value="([^"]+)"', resp.text)
|
|
if not m:
|
|
m = re.search(r'value="([^"]+)"\s+name="csrf_token"', resp.text)
|
|
assert m, f"csrf_token negasit in panoul Setari: {resp.text[:500]}"
|
|
return m.group(1)
|
|
|
|
|
|
def test_setari_subtaburi_prezente(client):
|
|
"""Panoul Setari expune cele 4 sub-taburi + forma de schimbare parola."""
|
|
_create_account_user("subtab@test.com")
|
|
_login(client, "subtab@test.com", "parolasecreta10")
|
|
|
|
resp = client.get("/?tab=cont")
|
|
assert resp.status_code == 200
|
|
html = resp.text
|
|
for eticheta in ("Date firma", "Credentiale RAR", "Cheie API", "Parola de acces"):
|
|
assert eticheta in html, f"Lipseste sub-tabul/zona '{eticheta}': {html[:800]}"
|
|
assert 'hx-post="/cont/schimba-parola"' in html, "Lipseste forma de schimbare parola"
|
|
|
|
|
|
def test_schimba_parola_succes(client):
|
|
"""Schimbare parola cu parola actuala corecta -> succes + login cu parola noua."""
|
|
_create_account_user("pwok@test.com", "parolaveche10")
|
|
_login(client, "pwok@test.com", "parolaveche10")
|
|
csrf = _get_csrf_from_cont(client)
|
|
|
|
resp = client.post("/cont/schimba-parola", data={
|
|
"csrf_token": csrf,
|
|
"parola_actuala": "parolaveche10",
|
|
"parola_noua": "parolanoua123",
|
|
"parola_confirma": "parolanoua123",
|
|
})
|
|
assert resp.status_code == 200, f"POST schimba-parola a returnat {resp.status_code}"
|
|
assert "schimbat" in resp.text.lower(), f"Lipseste confirmarea: {resp.text[:600]}"
|
|
|
|
# Parola noua verifica in DB
|
|
from app.db import get_connection
|
|
from app.users import verify_password
|
|
conn = get_connection()
|
|
try:
|
|
assert verify_password(conn, "pwok@test.com", "parolanoua123") is not None, \
|
|
"Parola noua nu autentifica"
|
|
assert verify_password(conn, "pwok@test.com", "parolaveche10") is None, \
|
|
"Parola veche inca autentifica dupa schimbare"
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def test_schimba_parola_actuala_gresita(client):
|
|
"""Parola actuala gresita -> eroare, parola neschimbata."""
|
|
_create_account_user("pwbad@test.com", "parolaveche10")
|
|
_login(client, "pwbad@test.com", "parolaveche10")
|
|
csrf = _get_csrf_from_cont(client)
|
|
|
|
resp = client.post("/cont/schimba-parola", data={
|
|
"csrf_token": csrf,
|
|
"parola_actuala": "gresita000",
|
|
"parola_noua": "parolanoua123",
|
|
"parola_confirma": "parolanoua123",
|
|
})
|
|
assert resp.status_code == 200
|
|
assert "incorect" in resp.text.lower(), f"Lipseste eroarea de parola gresita: {resp.text[:600]}"
|
|
|
|
from app.db import get_connection
|
|
from app.users import verify_password
|
|
conn = get_connection()
|
|
try:
|
|
assert verify_password(conn, "pwbad@test.com", "parolaveche10") is not None, \
|
|
"Parola veche trebuia sa ramana valida dupa esec"
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def test_schimba_parola_confirmare_nepotrivita(client):
|
|
"""Confirmarea nu se potriveste -> eroare, fara scriere."""
|
|
_create_account_user("pwmis@test.com", "parolaveche10")
|
|
_login(client, "pwmis@test.com", "parolaveche10")
|
|
csrf = _get_csrf_from_cont(client)
|
|
|
|
resp = client.post("/cont/schimba-parola", data={
|
|
"csrf_token": csrf,
|
|
"parola_actuala": "parolaveche10",
|
|
"parola_noua": "parolanoua123",
|
|
"parola_confirma": "altceva99999",
|
|
})
|
|
assert resp.status_code == 200
|
|
assert "confirmarea" in resp.text.lower() or "potriv" in resp.text.lower(), \
|
|
f"Lipseste eroarea de confirmare: {resp.text[:600]}"
|
|
|
|
|
|
def test_schimba_parola_prea_scurta(client):
|
|
"""Parola noua sub 10 caractere -> respinsa cu mesaj."""
|
|
_create_account_user("pwshort@test.com", "parolaveche10")
|
|
_login(client, "pwshort@test.com", "parolaveche10")
|
|
csrf = _get_csrf_from_cont(client)
|
|
|
|
resp = client.post("/cont/schimba-parola", data={
|
|
"csrf_token": csrf,
|
|
"parola_actuala": "parolaveche10",
|
|
"parola_noua": "scurt",
|
|
"parola_confirma": "scurt",
|
|
})
|
|
assert resp.status_code == 200
|
|
assert "scurt" in resp.text.lower(), f"Lipseste eroarea de lungime: {resp.text[:600]}"
|