feat(creds): livrare creds per-cerere la worker (criptat efemer + sesiuni per-cont)
Plan sect.5: parola RAR vine per-cerere, stocata CRIPTATA in submission pana la primul login reusit pe cont, apoi stearsa; JWT 30h acopera restul. - app/crypto.py: Fernet, cheie din AUTOPASS_creds_key (nesetata -> efemera la runtime, creds nu supravietuiesc restartului). encrypt/decrypt_creds. - schema + migrare: submissions.rar_creds_enc (creds criptate). - ingestie: cripteaza rar_credentials, le lipeste de fiecare submission nou. Niciodata in clar in DB. - worker: AccountSessions (login per-cont cu creds decriptate, cache JWT in memorie, sterge creds-urile contului dupa primul login + refresh nomenclator). 401 creds gresite -> error fara retry; token expirat -> invalidare + requeue; fara creds (restart) -> requeue "indisponibile" (ROAAUTO re-trimite). claim_one intoarce account_id + creds_enc; recover_orphans filtrabil pe cont. - requirements: cryptography==46.0.5. Nota: refresh nomenclator e acum lazy la primul login per-cont (nu la pornire); seed-ul fallback acopera editorul offline. 10 teste noi (tests/test_creds_delivery.py). 95 pass total. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Claude Agent
@@ -12,8 +12,14 @@ T2 implementat:
|
||||
- lease/timeout pe randuri 'sending' orfane.
|
||||
- re-login la token expirat (401 mid-sesiune) — JWT 30h, retry NU plafonat la 30h.
|
||||
|
||||
Ce NU e inca: livrarea creds per-cerere de la ROAAUTO (in schelet folosim creds <test>),
|
||||
criptare PII at-rest (P2), b64Image mare pe disc (P2).
|
||||
Creds per-cerere (plan sect. 5): fiecare submission poarta creds RAR CRIPTATE
|
||||
(rar_creds_enc). Worker-ul face login per CONT cu acele creds, cache-uieste JWT
|
||||
(30h) in memorie si STERGE creds-urile contului dupa primul login reusit. Token-ul
|
||||
in memorie acopera restul trimiterilor; la restart token-ul se pierde si contul
|
||||
re-loghează la urmatorul submission care aduce creds proaspete (degradare acceptata).
|
||||
Dev: `worker_use_test_creds` foloseste creds <test> cand submission-ul nu are enc.
|
||||
|
||||
Ce NU e inca: criptare PII payload at-rest (P2), b64Image mare pe disc (P2).
|
||||
|
||||
Pornire: python -m app.worker
|
||||
"""
|
||||
@@ -29,8 +35,9 @@ from datetime import datetime, timedelta, timezone
|
||||
import httpx
|
||||
|
||||
from ..config import Settings, get_settings, load_test_credentials
|
||||
from ..crypto import decrypt_creds
|
||||
from ..db import get_connection, init_db, write_heartbeat
|
||||
from ..mapping import upsert_nomenclator
|
||||
from ..mapping import DEFAULT_ACCOUNT_ID, upsert_nomenclator
|
||||
from ..payload import build_rar_payload
|
||||
from ..reconcile import match_finalizata
|
||||
from ..rar_client import RarAuthError, RarClient, RarError
|
||||
@@ -93,11 +100,15 @@ def requeue_with_backoff(conn, settings: Settings, submission_id: int, *, reason
|
||||
|
||||
|
||||
def claim_one(conn) -> dict | None:
|
||||
"""Claim atomic 'queued' -> 'sending', respectand next_attempt_at. Intoarce randul sau None."""
|
||||
"""Claim atomic 'queued' -> 'sending', respectand next_attempt_at. Intoarce randul sau None.
|
||||
|
||||
Randul include `account_id` si `rar_creds_enc` (creds RAR criptate) pentru
|
||||
login-ul per-cont din `run`.
|
||||
"""
|
||||
conn.execute("BEGIN IMMEDIATE")
|
||||
try:
|
||||
row = conn.execute(
|
||||
"SELECT id, payload_json FROM submissions WHERE status='queued' "
|
||||
"SELECT id, account_id, payload_json, rar_creds_enc FROM submissions WHERE status='queued' "
|
||||
"AND (next_attempt_at IS NULL OR next_attempt_at <= ?) ORDER BY id LIMIT 1",
|
||||
(_iso(_now()),),
|
||||
).fetchone()
|
||||
@@ -110,7 +121,12 @@ def claim_one(conn) -> dict | None:
|
||||
(row["id"],),
|
||||
)
|
||||
conn.execute("COMMIT")
|
||||
return {"id": row["id"], "content": json.loads(row["payload_json"])}
|
||||
return {
|
||||
"id": row["id"],
|
||||
"account_id": row["account_id"] if row["account_id"] is not None else DEFAULT_ACCOUNT_ID,
|
||||
"creds_enc": row["rar_creds_enc"],
|
||||
"content": json.loads(row["payload_json"]),
|
||||
}
|
||||
except Exception:
|
||||
conn.execute("ROLLBACK")
|
||||
raise
|
||||
@@ -174,14 +190,25 @@ def _handle_transient(conn, settings: Settings, rar: RarClient, token: str, sid:
|
||||
return "requeued"
|
||||
|
||||
|
||||
def recover_orphans(conn, settings: Settings, rar: RarClient, token: str) -> int:
|
||||
"""Randuri 'sending' mai vechi de lease (worker mort mid-POST). Reconciliaza; altfel requeue."""
|
||||
def recover_orphans(conn, settings: Settings, rar: RarClient, token: str, account_id: int | None = None) -> int:
|
||||
"""Randuri 'sending' mai vechi de lease (worker mort mid-POST). Reconciliaza; altfel requeue.
|
||||
|
||||
`account_id` filtreaza la orfanii unui cont (login-ul e per-cont); None = toti
|
||||
(compat teste / single-account).
|
||||
"""
|
||||
cutoff = _iso(_now() - timedelta(seconds=settings.worker_sending_lease_s))
|
||||
orphans = conn.execute(
|
||||
"SELECT id, payload_json FROM submissions WHERE status='sending' "
|
||||
"AND (sending_since IS NULL OR sending_since <= ?)",
|
||||
(cutoff,),
|
||||
).fetchall()
|
||||
if account_id is not None:
|
||||
orphans = conn.execute(
|
||||
"SELECT id, payload_json FROM submissions WHERE status='sending' "
|
||||
"AND (sending_since IS NULL OR sending_since <= ?) AND account_id=?",
|
||||
(cutoff, account_id),
|
||||
).fetchall()
|
||||
else:
|
||||
orphans = conn.execute(
|
||||
"SELECT id, payload_json FROM submissions WHERE status='sending' "
|
||||
"AND (sending_since IS NULL OR sending_since <= ?)",
|
||||
(cutoff,),
|
||||
).fetchall()
|
||||
recovered = 0
|
||||
for row in orphans:
|
||||
sid = row["id"]
|
||||
@@ -213,6 +240,69 @@ def _refresh_nomenclator(conn, rar: RarClient, token: str) -> None:
|
||||
print(f"[worker] nomenclator refresh esuat (continui): {exc}", flush=True)
|
||||
|
||||
|
||||
class AccountSessions:
|
||||
"""Sesiuni RAR per cont: login lazy cu creds din submission + cache JWT (30h).
|
||||
|
||||
La primul login reusit pentru un cont sterge creds-urile criptate ale contului
|
||||
(token-ul in memorie acopera restul). Pe 401 mid-sesiune se invalideaza sesiunea
|
||||
-> re-login la urmatorul submission cu creds.
|
||||
"""
|
||||
|
||||
def __init__(self, settings: Settings):
|
||||
self.settings = settings
|
||||
self._sessions: dict[int, tuple[RarClient, str]] = {}
|
||||
|
||||
def get_token(self, conn, account_id: int, creds: dict | None) -> str | None:
|
||||
"""Token valid pentru cont. Login daca lipseste din cache si avem creds; altfel None."""
|
||||
sess = self._sessions.get(account_id)
|
||||
if sess is not None:
|
||||
return sess[1]
|
||||
if not creds or not creds.get("email") or not creds.get("password"):
|
||||
return None
|
||||
rar = RarClient(self.settings)
|
||||
try:
|
||||
token = rar.login(creds["email"], creds["password"])
|
||||
except Exception:
|
||||
rar.close()
|
||||
raise
|
||||
self._sessions[account_id] = (rar, token)
|
||||
write_heartbeat(conn, rar_login_ok=True, detail=f"login RAR ok (cont {account_id})")
|
||||
# Creds nu mai sunt necesare: JWT acopera retry-urile -> sterge la rest.
|
||||
conn.execute(
|
||||
"UPDATE submissions SET rar_creds_enc=NULL WHERE account_id=? AND rar_creds_enc IS NOT NULL",
|
||||
(account_id,),
|
||||
)
|
||||
# Nomenclator live (autoritativ) la fiecare login proaspat.
|
||||
_refresh_nomenclator(conn, rar, token)
|
||||
return token
|
||||
|
||||
def rar(self, account_id: int) -> RarClient:
|
||||
return self._sessions[account_id][0]
|
||||
|
||||
def active(self) -> list[tuple[int, RarClient, str]]:
|
||||
return [(acct, rar, tok) for acct, (rar, tok) in self._sessions.items()]
|
||||
|
||||
def invalidate(self, account_id: int) -> None:
|
||||
sess = self._sessions.pop(account_id, None)
|
||||
if sess is not None:
|
||||
sess[0].close()
|
||||
|
||||
def close_all(self) -> None:
|
||||
for rar, _tok in self._sessions.values():
|
||||
rar.close()
|
||||
self._sessions.clear()
|
||||
|
||||
|
||||
def _creds_for(claimed: dict, settings: Settings) -> dict | None:
|
||||
"""Creds pentru un submission: decripteaza enc-ul lipit; altfel creds <test> (dev)."""
|
||||
creds = decrypt_creds(claimed.get("creds_enc"))
|
||||
if creds:
|
||||
return creds
|
||||
if settings.worker_use_test_creds:
|
||||
return load_test_credentials()
|
||||
return None
|
||||
|
||||
|
||||
def run() -> int:
|
||||
signal.signal(signal.SIGTERM, _stop)
|
||||
signal.signal(signal.SIGINT, _stop)
|
||||
@@ -222,9 +312,7 @@ def run() -> int:
|
||||
conn = get_connection()
|
||||
print(f"[worker] pornit (send_enabled={settings.worker_send_enabled}, env={settings.rar_env})", flush=True)
|
||||
|
||||
creds = load_test_credentials() if settings.worker_use_test_creds else None
|
||||
rar: RarClient | None = None
|
||||
token: str | None = None
|
||||
sessions = AccountSessions(settings)
|
||||
|
||||
while _running:
|
||||
try:
|
||||
@@ -233,38 +321,49 @@ def run() -> int:
|
||||
if not settings.worker_send_enabled:
|
||||
time.sleep(settings.worker_poll_interval_s)
|
||||
continue
|
||||
if not creds:
|
||||
time.sleep(settings.worker_poll_interval_s)
|
||||
continue
|
||||
|
||||
# Login lazy + token cache (JWT 30h).
|
||||
if rar is None or token is None:
|
||||
rar = RarClient(settings)
|
||||
token = rar.login(creds["email"], creds["password"])
|
||||
write_heartbeat(conn, rar_login_ok=True, detail="login RAR ok")
|
||||
# Refresh nomenclator live (autoritativ) la fiecare login proaspat —
|
||||
# alimenteaza fuzzy lookup-ul din editorul de mapari.
|
||||
_refresh_nomenclator(conn, rar, token)
|
||||
|
||||
recover_orphans(conn, settings, rar, token)
|
||||
|
||||
claimed = claim_one(conn)
|
||||
if claimed is None:
|
||||
# Nimic de trimis: recupereaza orfanii conturilor deja logate.
|
||||
for acct, rar, tok in sessions.active():
|
||||
recover_orphans(conn, settings, rar, tok, account_id=acct)
|
||||
time.sleep(settings.worker_poll_interval_s)
|
||||
continue
|
||||
|
||||
process_one(conn, settings, rar, token, claimed)
|
||||
sid = claimed["id"]
|
||||
account_id = claimed["account_id"]
|
||||
creds = _creds_for(claimed, settings)
|
||||
|
||||
try:
|
||||
token = sessions.get_token(conn, account_id, creds)
|
||||
except RarAuthError as exc:
|
||||
# Creds gresite (login 401): NU se face retry (plan, failure registry).
|
||||
mark(conn, sid, "error", rar_status_code=401, rar_error="credentiale RAR invalide")
|
||||
print(f"[worker] submission {sid} (cont {account_id}) -> error: {exc}", flush=True)
|
||||
continue
|
||||
|
||||
if token is None:
|
||||
# Fara creds disponibile (token pierdut la restart + creds sterse).
|
||||
# Re-pune in coada cu backoff; ROAAUTO re-trimite creds proaspete.
|
||||
requeue_with_backoff(conn, settings, sid, reason="creds RAR indisponibile (astept re-trimitere)")
|
||||
continue
|
||||
|
||||
rar = sessions.rar(account_id)
|
||||
# Recupereaza orfanii contului inainte de trimitere (acelasi token).
|
||||
recover_orphans(conn, settings, rar, token, account_id=account_id)
|
||||
try:
|
||||
process_one(conn, settings, rar, token, claimed)
|
||||
except RarAuthError as exc:
|
||||
# Token expirat mid-sesiune: invalideaza sesiunea, re-pune randul.
|
||||
print(f"[worker] cont {account_id} token expirat: {exc}; re-login data viitoare", flush=True)
|
||||
sessions.invalidate(account_id)
|
||||
requeue_with_backoff(conn, settings, sid, reason="token RAR expirat")
|
||||
|
||||
except RarAuthError as exc:
|
||||
print(f"[worker] login esuat / token expirat: {exc}", flush=True)
|
||||
token = None # forteaza re-login (acopera si expirarea JWT la 30h)
|
||||
time.sleep(settings.worker_poll_interval_s)
|
||||
except Exception as exc: # noqa: BLE001 — loop top-level: o eroare punctuala nu opreste worker-ul
|
||||
print(f"[worker] eroare neasteptata: {exc}", flush=True)
|
||||
time.sleep(settings.worker_poll_interval_s)
|
||||
|
||||
if rar is not None:
|
||||
rar.close()
|
||||
sessions.close_all()
|
||||
conn.close()
|
||||
print("[worker] oprit curat", flush=True)
|
||||
return 0
|
||||
|
||||
Reference in New Issue
Block a user