From cbcbb1995843ca1f09b6a7047908b1b7c3198df3 Mon Sep 17 00:00:00 2001 From: Sourav Chandhok <114119117+souravC01@users.noreply.github.com> Date: Fri, 11 Sep 2026 12:21:25 -0400 Subject: [PATCH 01/14] fix(print): Render cover letters as text --- frontend/src/components/CoverLetterModal.jsx | 66 ++++---------------- frontend/src/test/CoverLetterModal.test.jsx | 30 ++++++++- frontend/src/utils/printCoverLetter.js | 30 +++++++++ 3 files changed, 70 insertions(+), 56 deletions(-) create mode 100644 frontend/src/utils/printCoverLetter.js diff --git a/frontend/src/components/CoverLetterModal.jsx b/frontend/src/components/CoverLetterModal.jsx index 7c50930..6ba8673 100644 --- a/frontend/src/components/CoverLetterModal.jsx +++ b/frontend/src/components/CoverLetterModal.jsx @@ -1,6 +1,7 @@ import { useState, useEffect } from 'react'; import { Mail, Copy, Check, Sparkles, Loader2, MessageSquare, Send, FileText, Download } from 'lucide-react'; import { generateOutreach, exportCoverLetterDocx } from '../api/client'; +import { renderCoverLetter } from '../utils/printCoverLetter'; export default function CoverLetterModal({ isOpen, @@ -95,61 +96,16 @@ export default function CoverLetterModal({ alert('Please allow popups to download or print your PDF.'); return; } - const printCandidateName = candidateName; - const company = currentJob?.company || ''; - const role = currentJob?.title || ''; - const subject = data?.subject_line || ''; - const bodyHtml = data.cover_letter_pitch - .split('\n\n') - .map((p) => `

${p.replace(/\n/g, '
')}

`) - .join(''); - - printWindow.document.write(` - - - - Cover Letter - ${printCandidateName} - ${company} - - - -
-
${candidateName}
-
${new Date().toLocaleDateString('en-US', { year: 'numeric', month: 'long', day: 'numeric' })}
-
- Hiring Team
- ${company ? company + '
' : ''} -
- ${subject ? `
Subject: ${subject}
` : ''} -
-
- ${bodyHtml} -
- - -`); - printWindow.document.close(); + printWindow.opener = null; + renderCoverLetter(printWindow.document, { + candidateName, + company: currentJob?.company || '', + subject: data.subject_line || '', + body: data.cover_letter_pitch, + dateLabel: new Date().toLocaleDateString('en-US', { year: 'numeric', month: 'long', day: 'numeric' }), + }); + printWindow.focus(); + printWindow.print(); }; return ( diff --git a/frontend/src/test/CoverLetterModal.test.jsx b/frontend/src/test/CoverLetterModal.test.jsx index fa63bf9..7f2d07e 100644 --- a/frontend/src/test/CoverLetterModal.test.jsx +++ b/frontend/src/test/CoverLetterModal.test.jsx @@ -1,5 +1,5 @@ import { render, screen, waitFor, fireEvent, act } from '@testing-library/react' -import { describe, expect, it, vi, beforeEach } from 'vitest' +import { describe, expect, it, vi, beforeEach, afterEach } from 'vitest' import CoverLetterModal from '../components/CoverLetterModal' import * as apiClient from '../api/client' @@ -12,6 +12,7 @@ describe('CoverLetterModal 3-paragraph format and docx/pdf download', () => { beforeEach(() => { vi.clearAllMocks() }) + afterEach(() => vi.restoreAllMocks()) const mockJob = { title: 'Senior Software Engineer', @@ -42,6 +43,33 @@ describe('CoverLetterModal 3-paragraph format and docx/pdf download', () => { 'Hi! I noticed the Senior Software Engineer opening at Stripe and would love to connect.', } + it.each(['name', 'company', 'subject', 'body'])('prints %s as literal text without executable markup', async (field) => { + const hostile = '' + const popup = { document: document.implementation.createHTMLDocument(''), opener: window, print: vi.fn(), focus: vi.fn() } + vi.spyOn(window, 'open').mockReturnValue(popup) + apiClient.generateOutreach.mockResolvedValue({ ...mockOutreachData, + subject_line: field === 'subject' ? hostile : mockOutreachData.subject_line, + cover_letter_pitch: field === 'body' ? hostile : mockOutreachData.cover_letter_pitch, + }) + render() + fireEvent.click(await screen.findByRole('button', { name: /Download PDF/i })) + expect(popup.document.body.textContent).toContain(hostile) + expect(popup.document.querySelector('script,img,[onerror]')).toBeNull() + expect(popup.opener).toBeNull() + expect(popup.print).toHaveBeenCalledOnce() + }) + + it('explains when the browser blocks the print popup', async () => { + vi.spyOn(window, 'open').mockReturnValue(null) + const alert = vi.spyOn(window, 'alert').mockImplementation(() => {}) + apiClient.generateOutreach.mockResolvedValue(mockOutreachData) + render() + fireEvent.click(await screen.findByRole('button', { name: /Download PDF/i })) + expect(alert).toHaveBeenCalledWith(expect.stringMatching(/allow popups/i)) + }) + it('renders 3-paragraph cover letter, subject line, and linkedin note with docx and pdf download buttons', async () => { apiClient.generateOutreach.mockResolvedValue(mockOutreachData) diff --git a/frontend/src/utils/printCoverLetter.js b/frontend/src/utils/printCoverLetter.js new file mode 100644 index 0000000..945e0b0 --- /dev/null +++ b/frontend/src/utils/printCoverLetter.js @@ -0,0 +1,30 @@ +// Only static styles are interpreted; all document content is inserted as text. +export function renderCoverLetter(document, { candidateName, company, subject, body, dateLabel }) { + document.head.replaceChildren(); + document.body.replaceChildren(); + document.documentElement.lang = 'en'; + document.title = `Cover Letter - ${candidateName} - ${company}`; + const style = document.createElement('style'); + style.textContent = ` + @page { margin: 1in; size: letter; } + body { font-family: Arial, sans-serif; color: #1e293b; margin: 0; padding: 24px; font-size: 11pt; line-height: 1.6; } + .name { font-size: 16pt; font-weight: bold; margin-bottom: 4px; } + .date { font-size: 10pt; color: #64748b; margin-bottom: 16px; } + .recipient, .subject { margin-bottom: 16px; white-space: pre-wrap; } + .subject { font-weight: bold; } + p { margin: 0 0 14px; white-space: pre-wrap; overflow-wrap: anywhere; } + @media print { body { padding: 0; } } + `; + document.head.appendChild(style); + const append = (tag, value, className = '') => { + const node = document.createElement(tag); + node.textContent = String(value ?? ''); + node.className = className; + document.body.appendChild(node); + }; + append('div', candidateName, 'name'); + append('div', dateLabel, 'date'); + append('div', ['Hiring Team', company].filter(Boolean).join('\n'), 'recipient'); + if (subject) append('div', `Subject: ${subject}`, 'subject'); + String(body ?? '').split(/\r?\n\r?\n/).forEach(paragraph => append('p', paragraph)); +} From 35cd179d0413571751320109b999380de01c71de Mon Sep 17 00:00:00 2001 From: Sourav Chandhok <114119117+souravC01@users.noreply.github.com> Date: Fri, 11 Sep 2026 12:23:55 -0400 Subject: [PATCH 02/14] fix(auth): Require verified Google identity links --- backend/models.py | 1 + backend/routers/auth.py | 87 ++++++++++++++++-------- backend/storage.py | 14 ++-- frontend/src/api/client.js | 2 +- frontend/src/components/AuthModal.jsx | 5 ++ frontend/src/test/GooglePending.test.jsx | 14 ++++ requirements.txt | 1 + requirements/runtime.txt | 1 + tests/test_auth_api.py | 2 +- tests/test_google_verification.py | 35 ++++++++++ tests/test_identity_linking.py | 54 +++++++++++++++ 11 files changed, 180 insertions(+), 36 deletions(-) create mode 100644 frontend/src/test/GooglePending.test.jsx create mode 100644 tests/test_google_verification.py diff --git a/backend/models.py b/backend/models.py index 294032e..7f698ec 100644 --- a/backend/models.py +++ b/backend/models.py @@ -321,6 +321,7 @@ class AuthResponse(BaseModel): user: Optional[User] = None message: Optional[str] = None is_new_user: bool = False + requires_verification: bool = False class CoverLetterDocxRequest(BaseModel): diff --git a/backend/routers/auth.py b/backend/routers/auth.py index 947b018..2d6fad6 100644 --- a/backend/routers/auth.py +++ b/backend/routers/auth.py @@ -1,9 +1,10 @@ import hashlib -import json import os import re import secrets -import urllib.request +import requests +from google.auth.transport.requests import Request as GoogleRequest +from google.oauth2 import id_token from datetime import datetime, timedelta, timezone from typing import Any, Dict, Optional @@ -133,11 +134,39 @@ class GoogleAuthRequest(BaseModel): def verify_google_id_token(token: str) -> Dict[str, Any]: - """Validates Google ID token against tokeninfo endpoint and returns verified claims.""" - url = f"https://oauth2.googleapis.com/tokeninfo?id_token={token}" - req_obj = urllib.request.Request(url, headers={"User-Agent": "JobHelperGuru/1.0"}) - with urllib.request.urlopen(req_obj, timeout=5) as response: - return json.loads(response.read().decode("utf-8")) + """Verify signature, issuer, expiry and this application's audience.""" + audience = load_config().google_client_id + if not audience: + raise ValueError("Google sign-in is not configured") + with requests.Session() as session: + session.trust_env = False + request = GoogleRequest(session=session) + def bounded_request(url, method="GET", body=None, headers=None, **kwargs): + return request(url, method=method, body=body, headers=headers, timeout=5) + return id_token.verify_oauth2_token(token, bounded_request, audience=audience) + + +def google_is_authoritative(claims: Dict[str, Any]) -> bool: + email = str(claims.get("email", "")).lower() + return claims.get("email_verified") is True and ( + email.endswith("@gmail.com") or bool(claims.get("hd")) + ) + + +def google_verification_pending(user: User, is_new: bool = False) -> AuthResponse: + if is_new: + token = secrets.token_urlsafe(32) + get_storage().create_auth_token( + user.id, hashlib.sha256(token.encode()).hexdigest(), "verify_email", + (datetime.now(timezone.utc) + timedelta(minutes=30)).isoformat(), + ) + try: + email_service.send_verification(user.email, token, load_config().app_url) + except Exception: + # Tokens and transport credentials must never enter logs. + print("[WARN] Google account verification email delivery failed") + return AuthResponse(user=user, is_new_user=is_new, requires_verification=True, + message="Please verify your email before signing in. You can request a new link.") @router.post("/register", response_model=AuthResponse) @@ -406,18 +435,13 @@ def google_auth(req: GoogleAuthRequest): try: token_data = verify_google_id_token(token) - except Exception as e: - raise HTTPException(status_code=400, detail=f"Google token verification failed: {e}") + except Exception: + raise HTTPException(status_code=401, detail="Google token verification failed. Please sign in again.") cfg = load_config() - expected_aud = ( - cfg.google_client_id - or os.getenv("GOOGLE_CLIENT_ID") - or os.getenv("VITE_GOOGLE_CLIENT_ID") - or "999060759573-45b5m9cn9v7g6birnj9d8j65cqn72mfq.apps.googleusercontent.com" - ).strip() - token_aud = token_data.get("aud", "").strip() - if expected_aud and token_aud and token_aud != expected_aud: + expected_aud = (cfg.google_client_id or "").strip() + token_aud = token_data.get("aud") + if not expected_aud or token_aud != expected_aud: raise HTTPException( status_code=401, detail="Google token audience mismatch. Token was not issued for this application.", @@ -448,6 +472,8 @@ def google_auth(req: GoogleAuthRequest): existing_by_sub = storage.get_user_by_google_sub(sub) if existing_by_sub: user = storage.get_user_by_id(existing_by_sub["id"]) + if not user.email_verified: + return google_verification_pending(user) access_token = create_access_token(user.id, user.email, user.name, session_version=user.session_version) return AuthResponse(token=access_token, user=user, is_new_user=False) @@ -462,14 +488,18 @@ def google_auth(req: GoogleAuthRequest): detail="This email address is already bound to a different Google account.", ) + if not google_is_authoritative(token_data): + raise HTTPException(status_code=409, detail="Use password sign-in or password recovery for this account. Google cannot confirm current ownership of this email address.") + # Pre-account hijacking protection: if existing account was unverified, revoke password and sessions was_unverified = not bool(existing_by_email.get("email_verified")) - storage.link_google_identity( - user_id=existing_by_email["id"], - google_sub=sub, - clear_password=was_unverified, - verify_email=True, - ) + try: + storage.link_google_identity( + user_id=existing_by_email["id"], google_sub=sub, + clear_password=was_unverified, verify_email=True, + ) + except ValueError as exc: + raise HTTPException(status_code=409, detail=str(exc)) user = storage.get_user_by_id(existing_by_email["id"]) access_token = create_access_token(user.id, user.email, user.name, session_version=user.session_version) return AuthResponse(token=access_token, user=user, is_new_user=False) @@ -481,10 +511,12 @@ def google_auth(req: GoogleAuthRequest): name=name, avatar_url=picture, provider="google", - email_verified=True, + email_verified=google_is_authoritative(token_data), session_version=1, google_sub=sub, ) + if not user.email_verified: + return google_verification_pending(user, is_new=True) access_token = create_access_token(user.id, user.email, user.name, session_version=user.session_version) return AuthResponse(token=access_token, user=user, is_new_user=True) @@ -497,12 +529,7 @@ def me(current_user: User = Depends(get_current_user)): @router.get("/config") def get_auth_config(): cfg = load_config() - client_id = ( - cfg.google_client_id - or os.getenv("GOOGLE_CLIENT_ID") - or os.getenv("VITE_GOOGLE_CLIENT_ID") - or "999060759573-45b5m9cn9v7g6birnj9d8j65cqn72mfq.apps.googleusercontent.com" - ).strip() + client_id = (cfg.google_client_id or "").strip() return { "google_client_id": client_id, "require_email_verification": cfg.require_email_verification, diff --git a/backend/storage.py b/backend/storage.py index d535f8b..64b9896 100644 --- a/backend/storage.py +++ b/backend/storage.py @@ -445,19 +445,25 @@ def link_google_identity( clean_sub = google_sub.strip() with self._get_cursor() as cursor: if clear_password: + cursor.execute( + self._format_sql("UPDATE auth_tokens SET used_at = ? WHERE user_id = ? AND used_at IS NULL"), + (now, user_id), + ) cursor.execute( self._format_sql( - "UPDATE users SET google_sub = ?, hashed_password = NULL, email_verified = ?, session_version = session_version + 1, updated_at = ? WHERE id = ?" + "UPDATE users SET google_sub = ?, hashed_password = CASE WHEN email_verified = FALSE THEN NULL ELSE hashed_password END, email_verified = ?, session_version = session_version + 1, updated_at = ? WHERE id = ? AND (google_sub IS NULL OR google_sub = ?)" ), - (clean_sub, verify_email, now, user_id), + (clean_sub, verify_email, now, user_id, clean_sub), ) else: cursor.execute( self._format_sql( - "UPDATE users SET google_sub = ?, email_verified = ?, updated_at = ? WHERE id = ?" + "UPDATE users SET google_sub = ?, email_verified = ?, updated_at = ? WHERE id = ? AND (google_sub IS NULL OR google_sub = ?)" ), - (clean_sub, verify_email, now, user_id), + (clean_sub, verify_email, now, user_id, clean_sub), ) + if cursor.rowcount != 1: + raise ValueError("Google identity changed during sign-in. Please try again.") def increment_user_session_version(self, user_id: str) -> int: now = datetime.now().isoformat() diff --git a/frontend/src/api/client.js b/frontend/src/api/client.js index 32c3ed7..f47fe78 100644 --- a/frontend/src/api/client.js +++ b/frontend/src/api/client.js @@ -155,7 +155,7 @@ export async function googleAuthUser(credential) { throw new Error(data.detail || 'Google sign-in failed'); } const data = await res.json(); - setAuth(data.token, data.user); + if (data.token) setAuth(data.token, data.user); return data; } diff --git a/frontend/src/components/AuthModal.jsx b/frontend/src/components/AuthModal.jsx index 621ee3b..d873ba1 100644 --- a/frontend/src/components/AuthModal.jsx +++ b/frontend/src/components/AuthModal.jsx @@ -45,6 +45,11 @@ export default function AuthModal({ isOpen, onClose, onSuccess, initialMode = 'l setError(''); try { const data = await googleAuthUser(response.credential); + if (data.requires_verification) { + setEmail(data.user?.email || ''); + setMode('verify_pending'); + return; + } if (onSuccess) onSuccess(data.user, Boolean(data.is_new_user)); onClose(); } catch (err) { diff --git a/frontend/src/test/GooglePending.test.jsx b/frontend/src/test/GooglePending.test.jsx new file mode 100644 index 0000000..7c08cf8 --- /dev/null +++ b/frontend/src/test/GooglePending.test.jsx @@ -0,0 +1,14 @@ +import { it, expect, vi, afterEach } from 'vitest'; +import { googleAuthUser } from '../api/client'; + +afterEach(() => { vi.unstubAllGlobals(); localStorage.clear(); }); + +it('does not cache an unverified Google identity as a signed-in user', async () => { + vi.stubGlobal('fetch', vi.fn().mockResolvedValue({ ok: true, json: async () => ({ + token: null, requires_verification: true, user: { email: 'pending@example.test' }, + }) })); + const result = await googleAuthUser('synthetic'); + expect(result.requires_verification).toBe(true); + expect(localStorage.getItem('jh_user')).toBeNull(); + expect(localStorage.getItem('jh_token')).toBeNull(); +}); diff --git a/requirements.txt b/requirements.txt index d90cdc2..13fc9e1 100644 --- a/requirements.txt +++ b/requirements.txt @@ -6,6 +6,7 @@ psycopg2-binary>=2.9.9 boto3>=1.34.0 bcrypt>=4.2.0 pyjwt>=2.9.0 +google-auth>=2.40,<3 cryptography>=43.0.0 trafilatura>=1.12.0 beautifulsoup4>=4.12.3 diff --git a/requirements/runtime.txt b/requirements/runtime.txt index d90cdc2..13fc9e1 100644 --- a/requirements/runtime.txt +++ b/requirements/runtime.txt @@ -6,6 +6,7 @@ psycopg2-binary>=2.9.9 boto3>=1.34.0 bcrypt>=4.2.0 pyjwt>=2.9.0 +google-auth>=2.40,<3 cryptography>=43.0.0 trafilatura>=1.12.0 beautifulsoup4>=4.12.3 diff --git a/tests/test_auth_api.py b/tests/test_auth_api.py index 863352c..c16f82e 100644 --- a/tests/test_auth_api.py +++ b/tests/test_auth_api.py @@ -60,7 +60,7 @@ def test_google_auth_rejects_mismatched_audience(monkeypatch): mock_resp.read.return_value = json.dumps(fake_token_data).encode("utf-8") mock_resp.__enter__.return_value = mock_resp - with patch("urllib.request.urlopen", return_value=mock_resp): + with patch("backend.routers.auth.verify_google_id_token", return_value=fake_token_data): res = client.post("/api/auth/google", json={"credential": "fake-google-jwt"}) assert res.status_code == 401 assert "audience" in res.json()["detail"].lower() diff --git a/tests/test_google_verification.py b/tests/test_google_verification.py new file mode 100644 index 0000000..85bc3b3 --- /dev/null +++ b/tests/test_google_verification.py @@ -0,0 +1,35 @@ +import json +import time +from types import SimpleNamespace + +import jwt +import pytest +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric import rsa + +from backend.routers.auth import verify_google_id_token + + +@pytest.mark.parametrize('change', ['valid', 'expired', 'issuer', 'audience', 'missing_audience', 'signature']) +def test_google_library_verifies_signed_claims(monkeypatch, change): + key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + public = key.public_key().public_bytes(serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo).decode() + now = int(time.time()) + claims = dict(iss='https://accounts.google.com', aud='test-client-id', sub='synthetic-sub', + iat=now - 10, exp=now + 300, email='synthetic@gmail.com', email_verified=True) + if change == 'expired': claims['exp'] = now - 1 + if change == 'issuer': claims['iss'] = 'https://other.example.test' + if change == 'audience': claims['aud'] = 'other-client' + if change == 'missing_audience': del claims['aud'] + signing_key = rsa.generate_private_key(public_exponent=65537, key_size=2048) if change == 'signature' else key + token = jwt.encode(claims, signing_key, algorithm='RS256', headers={'kid': 'synthetic-key'}) + def certificates(self, url, **kwargs): + assert kwargs['timeout'] == 5 + assert 'token' not in url + return SimpleNamespace(status=200, data=json.dumps({'synthetic-key': public}).encode()) + monkeypatch.setattr('backend.routers.auth.GoogleRequest.__call__', certificates) + if change == 'valid': + assert verify_google_id_token(token)['sub'] == 'synthetic-sub' + else: + with pytest.raises(Exception): + verify_google_id_token(token) diff --git a/tests/test_identity_linking.py b/tests/test_identity_linking.py index 80072d7..61feb2f 100644 --- a/tests/test_identity_linking.py +++ b/tests/test_identity_linking.py @@ -53,6 +53,59 @@ def test_password_length_bounds(client): assert resp.status_code == 200 +def test_third_party_google_identity_cannot_link_by_email(client, monkeypatch): + from backend.routers.auth import get_storage + reg = client.post('/api/auth/register', json={ + 'email': 'existing@example.test', 'name': 'Owner', 'password': 'OwnerPassword123!', + }) + storage = get_storage() + uid = reg.json()['user']['id'] + storage.update_user_verification(uid, verified=True) + before = storage.get_user_by_email('existing@example.test') + monkeypatch.setattr('backend.routers.auth.verify_google_id_token', lambda _: { + 'email': 'existing@example.test', 'email_verified': True, + 'sub': 'unlinked-sub', 'aud': 'test-client-id', + }) + response = client.post('/api/auth/google', json={'credential': 'synthetic'}) + assert response.status_code == 409 + after = storage.get_user_by_email('existing@example.test') + for key in ('hashed_password', 'google_sub', 'email_verified', 'session_version'): + assert after[key] == before[key] + + +def test_google_missing_audience_is_rejected(client, monkeypatch): + monkeypatch.setattr('backend.routers.auth.verify_google_id_token', lambda _: { + 'email': 'test@gmail.com', 'email_verified': True, 'sub': 'no-aud', + }) + assert client.post('/api/auth/google', json={'credential': 'synthetic'}).status_code == 401 + + +def test_storage_cannot_overwrite_a_concurrently_linked_identity(client): + from backend.routers.auth import get_storage + storage = get_storage() + user = storage.create_user(email='bound@gmail.com', hashed_password=None, name='Bound', + google_sub='first-sub', email_verified=True) + with pytest.raises(ValueError): + storage.link_google_identity(user.id, 'second-sub', clear_password=True) + assert storage.get_user_by_id(user.id).google_sub == 'first-sub' + + +def test_new_third_party_google_requires_mailbox_proof(client, monkeypatch, fake_email_transport): + monkeypatch.setattr('backend.routers.auth.verify_google_id_token', lambda _: { + 'email': 'new@example.test', 'email_verified': True, 'sub': 'new-sub', 'aud': 'test-client-id', + }) + response = client.post('/api/auth/google', json={'credential': 'synthetic'}) + assert response.status_code == 200 + assert not response.json().get('token') + assert response.json()['requires_verification'] is True + token = fake_email_transport.get_last_verification_token('new@example.test') + assert token + # Repeated sign-in must not grant a session before the application challenge. + assert not client.post('/api/auth/google', json={'credential': 'synthetic'}).json().get('token') + assert client.post('/api/auth/verify-email/confirm', json={'token': token}).status_code == 200 + assert client.post('/api/auth/google', json={'credential': 'synthetic'}).json().get('token') + + def test_pre_account_hijacking_via_unverified_email_prevented(client, monkeypatch): # Step 1: Attacker registers unverified account for victim@example.test reg_resp = client.post( @@ -72,6 +125,7 @@ def test_pre_account_hijacking_via_unverified_email_prevented(client, monkeypatc "email": "victim@example.test", "email_verified": True, "sub": "google-victim-sub-9999", + "hd": "example.test", "name": "Real Victim", "aud": "test-client-id", } From 23753e7ac6ed6fd6b9576d4427abf004456f1e58 Mon Sep 17 00:00:00 2001 From: Sourav Chandhok <114119117+souravC01@users.noreply.github.com> Date: Fri, 11 Sep 2026 12:27:20 -0400 Subject: [PATCH 03/14] fix(network): Enforce safe outbound connections --- backend/services/ai_engine.py | 8 +- backend/services/outbound_http.py | 123 +++++++++-------------- backend/services/pinned_transport.py | 145 +++++++++++++++++++++++++++ backend/services/scraper.py | 24 ++--- requirements.txt | 4 +- requirements/runtime.txt | 4 +- tests/test_pinned_transport.py | 126 +++++++++++++++++++++++ tests/test_scraper.py | 7 +- 8 files changed, 344 insertions(+), 97 deletions(-) create mode 100644 backend/services/pinned_transport.py create mode 100644 tests/test_pinned_transport.py diff --git a/backend/services/ai_engine.py b/backend/services/ai_engine.py index 3d1c73d..06b9bfe 100644 --- a/backend/services/ai_engine.py +++ b/backend/services/ai_engine.py @@ -178,9 +178,13 @@ def _get_client(self) -> Optional[OpenAI]: local_ai_hosts=set(cfg.local_ai_hosts), is_ai_request=True, ) - validate_destination_url(self.api_base_url, policy) + validate_destination_url(self.api_base_url, policy, resolve_addresses=False) try: - return OpenAI(base_url=self.api_base_url, api_key=self.api_key, timeout=30.0) + import httpx2 + from backend.services.pinned_transport import PinnedTransport + http_client = httpx2.Client(transport=PinnedTransport(policy), follow_redirects=False, trust_env=False) + return OpenAI(base_url=self.api_base_url, api_key=self.api_key, timeout=30.0, + max_retries=0, http_client=http_client) except Exception as e: raise RuntimeError(f"Could not initialize AI Client for {self.model_name}: {str(e)}") from e diff --git a/backend/services/outbound_http.py b/backend/services/outbound_http.py index ee4338c..2c24acf 100644 --- a/backend/services/outbound_http.py +++ b/backend/services/outbound_http.py @@ -3,7 +3,11 @@ import urllib.parse from dataclasses import dataclass, field from typing import Optional, Set, Tuple, Any, Dict -import httpx +import httpx2 as httpx +import time +from contextlib import contextmanager +from contextvars import ContextVar +from dataclasses import replace class SSRFBlockedError(ValueError): @@ -23,6 +27,7 @@ def is_safe_ip(ip: ipaddress.IPv4Address | ipaddress.IPv6Address) -> bool: or ip.is_multicast or ip.is_reserved or ip.is_unspecified + or not ip.is_global ): return False @@ -53,7 +58,7 @@ class OutboundPolicy: max_body_bytes: int = 2 * 1024 * 1024 # 2 MiB -def validate_destination_url(url: str, policy: OutboundPolicy) -> Tuple[str, str]: +def validate_destination_url(url: str, policy: OutboundPolicy, *, resolve_addresses: bool = True) -> Tuple[str, str]: """ Enforces scheme, hostname, port, and resolved-address rules. Returns (clean_url, hostname). @@ -77,45 +82,39 @@ def validate_destination_url(url: str, policy: OutboundPolicy) -> Tuple[str, str port = parts.port or (443 if scheme == "https" else 80) - # 1. AI Request validation - if policy.is_ai_request: - if policy.mode == "local" and hostname in policy.local_ai_hosts: - # Local mode permits local AI endpoints on loopback - return clean, hostname - + local = policy.is_ai_request and policy.mode == "local" and hostname in policy.local_ai_hosts + if not local and port != (443 if scheme == "https" else 80): + raise SSRFBlockedError("Destination port is not permitted") + if policy.is_ai_request and not local: if scheme != "https": raise SSRFBlockedError("Cloud AI provider endpoints must use HTTPS.") - - if policy.allowed_provider_hosts and hostname not in policy.allowed_provider_hosts: + if not policy.allowed_provider_hosts or hostname not in policy.allowed_provider_hosts: raise SSRFBlockedError(f"Host '{hostname}' is not in allowed AI provider destinations.") - - return clean, hostname - - # 2. General Scraping / Web Request validation - # Check if hostname is direct IP literal try: - direct_ip = ipaddress.ip_address(hostname) - if not is_safe_ip(direct_ip): - raise SSRFBlockedError(f"Destination IP '{hostname}' is private or disallowed.") - return clean, hostname + literal = ipaddress.ip_address(hostname) except ValueError: - pass + literal = None + if literal is not None and not (literal.is_loopback if local else is_safe_ip(literal)): + raise SSRFBlockedError("Destination IP is private or disallowed") + if resolve_addresses: + from backend.services.pinned_transport import resolve_public_addresses + resolve_public_addresses(hostname, port, allow_loopback=local) + return clean, hostname - # Resolve DNS and check all returned IP addresses - try: - resolved = socket.getaddrinfo(hostname, port, type=socket.SOCK_STREAM) - for entry in resolved: - sockaddr = entry[4] - ip_str = sockaddr[0] - ip_obj = ipaddress.ip_address(ip_str) - if not is_safe_ip(ip_obj): - raise SSRFBlockedError(f"Host '{hostname}' resolves to private or disallowed address: {ip_str}") - except socket.gaierror as e: - raise SSRFBlockedError(f"Failed to resolve host '{hostname}': {e}") - return clean, hostname +_operation_deadline = ContextVar("outbound_operation_deadline", default=None) +@contextmanager +def outbound_budget(seconds=30): + existing = _operation_deadline.get() + deadline = min(existing, time.monotonic() + seconds) if existing else time.monotonic() + seconds + token = _operation_deadline.set(deadline) + try: + yield + finally: + _operation_deadline.reset(token) + class SafeHttpClient: """HTTP client enforcing outbound request policies, redirect bounds, and body limits.""" @@ -133,50 +132,28 @@ def request( headers: Optional[Dict[str, str]] = None, **kwargs, ) -> httpx.Response: + from backend.services.pinned_transport import PinnedTransport, remaining active_policy = policy or self.default_policy - limit_bytes = max_bytes or active_policy.max_body_bytes - + if max_bytes is not None: + active_policy = replace(active_policy, max_body_bytes=min(max_bytes, active_policy.max_body_bytes)) + deadline = _operation_deadline.get() or time.monotonic() + min(timeout, 30) current_url = url req_headers = dict(headers or {}) - hops = 0 - - with httpx.Client(follow_redirects=False, timeout=timeout) as client: - while hops <= active_policy.max_redirects: - clean_url, host = validate_destination_url(current_url, active_policy) - - resp = client.request( - method, - clean_url, - headers=req_headers, - **kwargs, - ) - - # Check for redirect status codes - if resp.is_redirect and "location" in resp.headers: - hops += 1 - location = resp.headers["location"].strip() - next_url = urllib.parse.urljoin(clean_url, location) - - # Strip Authorization if redirecting across different hostnames or downgrading to HTTP - next_parts = urllib.parse.urlsplit(next_url) - curr_parts = urllib.parse.urlsplit(clean_url) - - if next_parts.hostname != curr_parts.hostname or next_parts.scheme != curr_parts.scheme: - req_headers.pop("Authorization", None) - req_headers.pop("authorization", None) - - current_url = next_url - continue - - # Enforce response body size limit - content = resp.content - if len(content) > limit_bytes: - raise ValueError(f"Response body exceeded limit of {limit_bytes} bytes.") - - return resp - - raise SSRFBlockedError(f"Too many redirects (exceeded limit of {active_policy.max_redirects}).") - + with httpx.Client(transport=PinnedTransport(active_policy), follow_redirects=False, trust_env=False) as client: + for hops in range(active_policy.max_redirects + 1): + clean_url, _ = validate_destination_url(current_url, active_policy, resolve_addresses=False) + resp = client.request(method, clean_url, headers=req_headers, timeout=remaining(deadline), + extensions={"jhg_deadline": deadline}, **kwargs) + if not resp.is_redirect or "location" not in resp.headers: + return resp + if hops == active_policy.max_redirects: + raise SSRFBlockedError("Too many redirects") + next_url = urllib.parse.urljoin(clean_url, resp.headers["location"]) + previous = urllib.parse.urlsplit(clean_url) + following = urllib.parse.urlsplit(next_url) + if (previous.scheme, previous.netloc) != (following.scheme, following.netloc): + req_headers = {k: v for k, v in req_headers.items() if k.lower() not in {"authorization", "cookie"}} + current_url = next_url def get(self, url: str, **kwargs) -> httpx.Response: return self.request("GET", url, **kwargs) diff --git a/backend/services/pinned_transport.py b/backend/services/pinned_transport.py new file mode 100644 index 0000000..55a45da --- /dev/null +++ b/backend/services/pinned_transport.py @@ -0,0 +1,145 @@ +"""Connection enforcement for the pinned httpx2/httpcore2 transport versions. + +DNS results are checked at the dial boundary. TLS still sees the original host. +The pool adapter is version-pinned and covered by transport integration tests. +""" +import ipaddress +import socket +import time +import zlib +from concurrent.futures import ThreadPoolExecutor +from contextvars import ContextVar +from threading import BoundedSemaphore + +import httpcore2 +import httpx2 + +from backend.services.outbound_http import OutboundPolicy, SSRFBlockedError, is_safe_ip_string, validate_destination_url + +_deadline = ContextVar('outbound_deadline', default=None) +_dns_workers = ThreadPoolExecutor(max_workers=8, thread_name_prefix='outbound-dns') +_dns_slots = BoundedSemaphore(16) + + +def remaining(deadline, timeout=None): + budget = deadline - time.monotonic() + if budget <= 0: + raise TimeoutError('Outbound request deadline exceeded') + return min(budget, timeout) if timeout is not None else budget + + +def resolve_public_addresses(host, port, *, allow_loopback=False, deadline=None): + deadline = deadline or time.monotonic() + 5 + try: + addresses = (str(ipaddress.ip_address(host)),) + except ValueError: + if not _dns_slots.acquire(blocking=False): + raise TimeoutError('DNS resolver capacity exhausted') + future = _dns_workers.submit(socket.getaddrinfo, host, port, type=socket.SOCK_STREAM) + future.add_done_callback(lambda _: _dns_slots.release()) + rows = future.result(timeout=remaining(deadline, 5)) + addresses = tuple(dict.fromkeys(row[4][0] for row in rows)) + if not addresses: + raise SSRFBlockedError('DNS returned no destination addresses') + for address in addresses: + allowed = ipaddress.ip_address(address).is_loopback if allow_loopback else is_safe_ip_string(address) + if not allowed: + raise SSRFBlockedError('Destination resolves to a private or disallowed address') + return addresses + + +class DeadlineStream(httpcore2.NetworkStream): + def __init__(self, stream, deadline): + self.stream, self.deadline = stream, deadline + + def read(self, max_bytes, timeout=None): + data = self.stream.read(min(max_bytes, 65536), timeout=remaining(self.deadline, timeout)) + remaining(self.deadline) + return data + + def write(self, buffer, timeout=None): + self.stream.write(buffer, timeout=remaining(self.deadline, timeout)) + remaining(self.deadline) + + def start_tls(self, ssl_context, server_hostname=None, timeout=None): + stream = self.stream.start_tls(ssl_context=ssl_context, server_hostname=server_hostname, + timeout=remaining(self.deadline, timeout)) + return DeadlineStream(stream, self.deadline) + + def close(self): + self.stream.close() + + def get_extra_info(self, info): + return self.stream.get_extra_info(info) + + +class PinnedNetworkBackend(httpcore2.SyncBackend): + def __init__(self, policy): + self.policy = policy + + def connect_tcp(self, host, port, timeout=None, local_address=None, socket_options=None): + deadline = _deadline.get() or time.monotonic() + min(timeout or 30, 30) + local = self.policy.mode == 'local' and self.policy.is_ai_request and host in self.policy.local_ai_hosts + addresses = resolve_public_addresses(host, port, allow_loopback=local, deadline=deadline) + last_error = None + for address in addresses: + try: + stream = super().connect_tcp(host=address, port=port, timeout=remaining(deadline, timeout), + local_address=local_address, socket_options=socket_options) + return DeadlineStream(stream, deadline) + except (OSError, httpcore2.ConnectError) as exc: + last_error = exc + raise last_error + + +class PinnedTransport(httpx2.HTTPTransport): + def __init__(self, policy: OutboundPolicy): + # HTTPTransport.handle_request converts between HTTPX and HTTP Core models. + # Supply its pool with our connection backend; never create an ambient proxy. + self.policy = policy + self._pool = httpcore2.ConnectionPool( + network_backend=PinnedNetworkBackend(policy), max_connections=8, + max_keepalive_connections=0, retries=0, + ) + + def handle_request(self, request): + deadline = request.extensions.get('jhg_deadline', time.monotonic() + 30) + remaining(deadline) + validate_destination_url(str(request.url), self.policy, resolve_addresses=False) + request.headers['accept-encoding'] = 'identity' + token = _deadline.set(deadline) + try: + response = super().handle_request(request) + try: + if response.is_redirect: + if self.policy.is_ai_request: + raise SSRFBlockedError('AI provider redirects are not permitted') + return httpx2.Response(response.status_code, headers=response.headers, content=b'') + encoding = response.headers.get('content-encoding', 'identity').lower() + if encoding not in ('identity', 'gzip', 'deflate'): + raise ValueError('Unsupported response content encoding') + decoder = zlib.decompressobj(31 if encoding == 'gzip' else 15) if encoding != 'identity' else None + body = bytearray() + encoded = 0 + limit = self.policy.max_body_bytes + for chunk in response.iter_raw(): + remaining(deadline) + encoded += len(chunk) + if encoded > limit: + raise ValueError('Encoded response exceeds limit') + data = decoder.decompress(chunk, limit - len(body) + 1) if decoder else chunk + body.extend(data) + if len(body) > limit or (decoder and decoder.unconsumed_tail): + raise ValueError('Decoded response exceeds limit') + if decoder and decoder.unused_data: + raise ValueError('Multiple compressed response streams are not supported') + if decoder and not decoder.eof: + raise ValueError('Incomplete compressed response') + remaining(deadline) + headers = [(key, val) for key, val in response.headers.multi_items() + if key.lower() not in {'content-encoding', 'content-length', 'transfer-encoding'}] + return httpx2.Response(response.status_code, headers=headers, content=bytes(body)) + finally: + response.close() + finally: + _deadline.reset(token) diff --git a/backend/services/scraper.py b/backend/services/scraper.py index 9126246..aaf80a6 100644 --- a/backend/services/scraper.py +++ b/backend/services/scraper.py @@ -50,6 +50,11 @@ def __init__(self): }) def scrape_url(self, url: str, timeout: int = 15) -> ScrapedJob: + from backend.services.outbound_http import outbound_budget + with outbound_budget(30): + return self._scrape_url(url, timeout) + + def _scrape_url(self, url: str, timeout: int = 15) -> ScrapedJob: clean_url = url.strip() if not clean_url.startswith("http://") and not clean_url.startswith("https://"): clean_url = "https://" + clean_url @@ -63,21 +68,6 @@ def scrape_url(self, url: str, timeout: int = 15) -> ScrapedJob: html = None - # 1. Try TLS browser impersonation via curl_cffi for anti-bot protected sites (Indeed, Glassdoor, Cloudflare) - if CURL_CFFI_AVAILABLE and is_safe_url(clean_url): - for profile in ("safari17_0", "safari15_5", "chrome124"): - try: - cffi_resp = cffi_requests.get(clean_url, impersonate=profile, timeout=timeout, allow_redirects=False) - if ( - cffi_resp.status_code == 200 - and "Authenticating..." not in cffi_resp.text - and "Security Check" not in cffi_resp.text - ): - html = cffi_resp.text - break - except Exception: - continue - # 2. Fallback to SafeHttpClient with redirect and body limits if not html: try: @@ -562,8 +552,8 @@ def _try_fetch_workday_cxs(self, url: str, timeout: int = 15) -> Optional[Scrape cxs_url = f"https://{host}/wday/cxs/{tenant}/{portal}/job/{job_slug}" try: - resp = self.session.get(cxs_url, headers={"Accept": "application/json"}, timeout=timeout) - if resp.ok: + resp = self.safe_client.get(cxs_url, headers={"Accept": "application/json"}, timeout=timeout) + if resp.is_success: data = resp.json() info = data.get("jobPostingInfo", {}) title = info.get("title", "") diff --git a/requirements.txt b/requirements.txt index 13fc9e1..5322b23 100644 --- a/requirements.txt +++ b/requirements.txt @@ -11,7 +11,9 @@ cryptography>=43.0.0 trafilatura>=1.12.0 beautifulsoup4>=4.12.3 requests>=2.32.0 -openai>=1.40.0 +openai==3.8.0 +httpx2==2.12.0 +httpcore2==2.12.0 pypdf>=5.0.0 python-docx>=1.1.0 openpyxl>=3.1.5 diff --git a/requirements/runtime.txt b/requirements/runtime.txt index 13fc9e1..5322b23 100644 --- a/requirements/runtime.txt +++ b/requirements/runtime.txt @@ -11,7 +11,9 @@ cryptography>=43.0.0 trafilatura>=1.12.0 beautifulsoup4>=4.12.3 requests>=2.32.0 -openai>=1.40.0 +openai==3.8.0 +httpx2==2.12.0 +httpcore2==2.12.0 pypdf>=5.0.0 python-docx>=1.1.0 openpyxl>=3.1.5 diff --git a/tests/test_pinned_transport.py b/tests/test_pinned_transport.py new file mode 100644 index 0000000..363c1f7 --- /dev/null +++ b/tests/test_pinned_transport.py @@ -0,0 +1,126 @@ +import gzip +import socket +import time +from unittest.mock import Mock + +import httpx2 +import httpcore2 +import pytest + +from backend.services.outbound_http import OutboundPolicy, SSRFBlockedError, is_safe_ip_string + + +def test_shared_address_space_is_not_public(): + assert not is_safe_ip_string('100.64.0.1') + + +def test_connection_revalidates_dns_and_never_dials_private(monkeypatch): + from backend.services.pinned_transport import PinnedNetworkBackend + dial = Mock() + monkeypatch.setattr('httpcore2.SyncBackend.connect_tcp', dial) + monkeypatch.setattr(socket, 'getaddrinfo', lambda *a, **k: [(socket.AF_INET, socket.SOCK_STREAM, 6, '', ('127.0.0.1', 443))]) + backend = PinnedNetworkBackend(OutboundPolicy(is_scraping_request=True)) + with pytest.raises(SSRFBlockedError): + backend.connect_tcp('rebind.example.test', 443, timeout=1) + dial.assert_not_called() + + +def test_connection_dials_validated_literal_and_preserves_tls_name(monkeypatch): + from backend.services.pinned_transport import PinnedNetworkBackend + stream = Mock() + stream.start_tls.return_value = stream + dial = Mock(return_value=stream) + monkeypatch.setattr('httpcore2.SyncBackend.connect_tcp', dial) + monkeypatch.setattr(socket, 'getaddrinfo', lambda *a, **k: [(socket.AF_INET, socket.SOCK_STREAM, 6, '', ('8.8.8.8', 443))]) + wrapped = PinnedNetworkBackend(OutboundPolicy()).connect_tcp('public.example.test', 443, timeout=1) + assert dial.call_args.kwargs['host'] == '8.8.8.8' + wrapped.start_tls(ssl_context=Mock(), server_hostname='public.example.test', timeout=1) + assert stream.start_tls.call_args.kwargs['server_hostname'] == 'public.example.test' + + +@pytest.mark.parametrize('compressed', [False, True]) +def test_transport_stops_oversize_body(monkeypatch, compressed): + from backend.services.pinned_transport import PinnedTransport + data = gzip.compress(b'x' * 10000) if compressed else b'x' * 10000 + closed = [] + class Stream(httpx2.SyncByteStream): + def __iter__(self): + yield data + pytest.fail('oversized response must stop before reading the next chunk') + def close(self): closed.append(True) + monkeypatch.setattr(httpx2.HTTPTransport, 'handle_request', lambda self, req: httpx2.Response( + 200, headers={'content-encoding': 'gzip'} if compressed else {}, stream=Stream())) + transport = PinnedTransport(OutboundPolicy(max_body_bytes=100)) + with pytest.raises(ValueError, match='limit'): + transport.handle_request(httpx2.Request('GET', 'https://example.test')) + assert closed + transport.close() + + +def test_ai_redirect_is_never_followed(monkeypatch): + from backend.services.ai_engine import AIEngine + requests = [] + def reply(self, request): + requests.append(str(request.url)) + return httpx2.Response(307, headers={'location': 'https://127.0.0.1/internal'}, stream=httpx2.ByteStream(b'')) + monkeypatch.setattr(httpx2.HTTPTransport, 'handle_request', reply) + client = AIEngine(api_key='synthetic', model_name='test')._get_client() + try: + with pytest.raises(Exception): + client.chat.completions.create(model='test', messages=[{'role': 'user', 'content': 'synthetic'}]) + assert len(requests) == 1 + finally: + client.close() + + +def test_expired_deadline_stops_before_connect(monkeypatch): + from backend.services.pinned_transport import PinnedTransport + send = Mock() + monkeypatch.setattr(httpx2.HTTPTransport, 'handle_request', send) + transport = PinnedTransport(OutboundPolicy()) + request = httpx2.Request('GET', 'https://example.test', extensions={'jhg_deadline': time.monotonic() - 1}) + with pytest.raises(TimeoutError): + transport.handle_request(request) + send.assert_not_called() + transport.close() + + +def test_real_pool_uses_pinned_backend_and_original_tls_hostname(monkeypatch): + from backend.services.pinned_transport import PinnedTransport + tls_names = [] + class Stream(httpcore2.MockStream): + def start_tls(self, ssl_context, server_hostname=None, timeout=None): + assert ssl_context.check_hostname + tls_names.append(server_hostname) + return self + stream = Stream([b'HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nok']) + dial = Mock(return_value=stream) + monkeypatch.setattr(httpcore2.SyncBackend, 'connect_tcp', dial) + monkeypatch.setattr(socket, 'getaddrinfo', lambda *a, **k: [(socket.AF_INET, socket.SOCK_STREAM, 6, '', ('8.8.8.8', 443))]) + with httpx2.Client(transport=PinnedTransport(OutboundPolicy()), trust_env=False) as client: + assert client.get('https://public.example.test').text == 'ok' + assert dial.call_args.kwargs['host'] == '8.8.8.8' + assert tls_names == ['public.example.test'] + + +def test_mixed_dns_addresses_block_entire_connection(monkeypatch): + from backend.services.pinned_transport import PinnedNetworkBackend + dial = Mock() + monkeypatch.setattr(httpcore2.SyncBackend, 'connect_tcp', dial) + monkeypatch.setattr(socket, 'getaddrinfo', lambda *a, **k: [ + (socket.AF_INET, socket.SOCK_STREAM, 6, '', (address, 443)) + for address in ('8.8.8.8', '10.0.0.1')]) + with pytest.raises(SSRFBlockedError): + PinnedNetworkBackend(OutboundPolicy()).connect_tcp('mixed.example.test', 443) + dial.assert_not_called() + + +@pytest.mark.parametrize('url', ['https://example.test:8443', 'http://example.test:8080', 'http://100.64.0.1']) +def test_ports_and_shared_address_space_blocked_before_send(monkeypatch, url): + from backend.services.pinned_transport import PinnedTransport + send = Mock() + monkeypatch.setattr(httpx2.HTTPTransport, 'handle_request', send) + with PinnedTransport(OutboundPolicy()) as transport: + with pytest.raises(SSRFBlockedError): + transport.handle_request(httpx2.Request('GET', url)) + send.assert_not_called() diff --git a/tests/test_scraper.py b/tests/test_scraper.py index 0866fd9..64e01a4 100644 --- a/tests/test_scraper.py +++ b/tests/test_scraper.py @@ -197,7 +197,7 @@ def test_scrape_url_blocks_internal_and_cloud_metadata(): assert any(term in job.raw_text.lower() for term in ["blocked", "disallowed", "invalid", "security"]) -def test_scrape_indeed_url_with_tls_impersonation(monkeypatch): +def test_scrape_captured_job_uses_protected_transport(monkeypatch): scraper = ScraperService() captured_html = """ Software Engineering Developer - Example Co @@ -212,8 +212,9 @@ class CapturedResponse: monkeypatch.setattr("socket.getaddrinfo", lambda *args, **kwargs: [(None, None, None, None, ("93.184.216.34", 0))]) monkeypatch.setattr("backend.services.scraper.CURL_CFFI_AVAILABLE", True) - monkeypatch.setattr("backend.services.scraper.cffi_requests.get", lambda *args, **kwargs: CapturedResponse()) - monkeypatch.setattr(scraper.session, "get", lambda *args, **kwargs: pytest.fail("standard transport must not be used")) + monkeypatch.setattr("backend.services.scraper.cffi_requests.get", lambda *args, **kwargs: pytest.fail("unprotected curl must not be used")) + monkeypatch.setattr(scraper.safe_client, "get", lambda *args, **kwargs: CapturedResponse()) + monkeypatch.setattr(scraper.session, "get", lambda *args, **kwargs: pytest.fail("unprotected requests must not be used")) job = scraper.scrape_url("https://ca.indeed.com/viewjob?jk=d721d1b5ad371161&from=shareddesktop_copy") assert "Software Engineering" in job.title assert "Java" in job.raw_text From 6c85079a6056823cb4a7f65454c49f7fa92cefb2 Mon Sep 17 00:00:00 2001 From: Sourav Chandhok <114119117+souravC01@users.noreply.github.com> Date: Fri, 11 Sep 2026 23:45:35 -0400 Subject: [PATCH 04/14] fix(limits): Enforce atomic rate limits and concurrency leases --- backend/main.py | 180 +++++++++++++++------------- backend/routers/auth.py | 14 ++- backend/services/rate_limiter.py | 120 ++++++++++++------- backend/services/resource_leases.py | 90 ++++++++++++++ backend/storage.py | 35 ++++++ tests/test_atomic_limits.py | 50 ++++++++ tests/test_resource_leases.py | 72 +++++++++++ 7 files changed, 436 insertions(+), 125 deletions(-) create mode 100644 backend/services/resource_leases.py create mode 100644 tests/test_atomic_limits.py create mode 100644 tests/test_resource_leases.py diff --git a/backend/main.py b/backend/main.py index 4455aa2..51982a0 100644 --- a/backend/main.py +++ b/backend/main.py @@ -12,6 +12,7 @@ from backend.config import load_config from backend.services.rate_limiter import RateLimiter, get_client_ip +from backend.services.resource_leases import ResourceLeases from backend.models import ( Application, ApplicationCreate, @@ -227,8 +228,16 @@ def analyze_job( if not req.url and not job_input_text: raise HTTPException(status_code=400, detail="Either a URL or job text must be provided.") - if not current_user: - limiter = RateLimiter(storage) + limiter = RateLimiter(storage) + if current_user: + allowed, _, retry_after = limiter.check("user_ai_ops", current_user.id, limit=30, window_seconds=60) + if not allowed: + raise HTTPException( + status_code=429, + detail="Too many analysis requests. Please try again later.", + headers={"Retry-After": str(int(retry_after))}, + ) + else: cfg = load_config() client_ip = get_client_ip(request, trust_proxy_headers=getattr(cfg, "trust_proxy_headers", False)) allowed, _, retry_after = limiter.check("anon_analysis", client_ip, limit=10, window_seconds=60) @@ -239,91 +248,100 @@ def analyze_job( headers={"Retry-After": str(int(retry_after))}, ) - scraped_title = "" - scraped_company = "" - scraped_location = "" - job_text = "" - source_url = req.url or "" - - if req.url: - scraped = scraper.scrape_url(req.url) - scraped_title = scraped.title - scraped_company = scraped.company - scraped_location = scraped.location - job_text = scraped.raw_text - elif job_input_text: - parsed = scraper.parse_raw_text(job_input_text) - scraped_title = parsed.title - scraped_company = parsed.company - scraped_location = parsed.location - job_text = parsed.raw_text - - if ( - not job_text - or len(job_text.strip()) < 30 - or job_text.strip().startswith("Error fetching URL:") - or job_text.strip().startswith("Disallowed or private URL target") - ): - raise HTTPException( - status_code=400, - detail=( - "Could not extract readable text from this job URL (it may require login or block automated scraping). " - "Please copy and paste the job description text directly into the 'Paste Job Text' tab." - ), - ) + cfg = load_config() + client_ip = get_client_ip(request, trust_proxy_headers=getattr(cfg, "trust_proxy_headers", False)) + lease_service = ResourceLeases(storage) + lease_subject = current_user.id if current_user else f"anon:{client_ip}" + lease_id = lease_service.acquire(user_id=lease_subject, kind="analysis", ttl_seconds=30) - user_id = current_user.id if current_user else None - ai = get_ai_engine(user_id=user_id) try: - analysis = ai.analyze_job(job_text, source_url=source_url) - except Exception as e: - raise HTTPException( - status_code=502, - detail={ - "message": str(e), - "can_switch_offline": True, - "error_type": "ai_api_error", - "model_name": ai.model_name, - }, - ) + scraped_title = "" + scraped_company = "" + scraped_location = "" + job_text = "" + source_url = req.url or "" + + if req.url: + scraped = scraper.scrape_url(req.url) + scraped_title = scraped.title + scraped_company = scraped.company + scraped_location = scraped.location + job_text = scraped.raw_text + elif job_input_text: + parsed = scraper.parse_raw_text(job_input_text) + scraped_title = parsed.title + scraped_company = parsed.company + scraped_location = parsed.location + job_text = parsed.raw_text + + if ( + not job_text + or len(job_text.strip()) < 30 + or job_text.strip().startswith("Error fetching URL:") + or job_text.strip().startswith("Disallowed or private URL target") + ): + raise HTTPException( + status_code=400, + detail=( + "Could not extract readable text from this job URL (it may require login or block automated scraping). " + "Please copy and paste the job description text directly into the 'Paste Job Text' tab." + ), + ) - UNKNOWN_TITLES = {"", "Open Position", "Detected Role", "Unknown Role", "Role Title", "Exact Role Title"} - UNKNOWN_COMPANIES = {"", "Unknown Company", "Company", "Company Name", "Detected Company"} - UNKNOWN_LOCATIONS = {"", "Unknown", "Identified Location", "City, State or Remote/Hybrid", "Unknown Location"} + user_id = current_user.id if current_user else None + ai = get_ai_engine(user_id=user_id) + try: + analysis = ai.analyze_job(job_text, source_url=source_url) + except Exception as e: + raise HTTPException( + status_code=502, + detail={ + "message": str(e), + "can_switch_offline": True, + "error_type": "ai_api_error", + "model_name": ai.model_name, + }, + ) - # Prefer scraped title/company/location if AI/heuristic returned generic placeholders - if (not analysis.title or analysis.title in UNKNOWN_TITLES) and scraped_title and scraped_title not in UNKNOWN_TITLES: - analysis.title = scraped_title - if (not analysis.company or analysis.company in UNKNOWN_COMPANIES) and scraped_company and scraped_company not in UNKNOWN_COMPANIES: - analysis.company = scraped_company - if (not analysis.location or analysis.location in UNKNOWN_LOCATIONS) and scraped_location and scraped_location not in UNKNOWN_LOCATIONS: - analysis.location = scraped_location + UNKNOWN_TITLES = {"", "Open Position", "Detected Role", "Unknown Role", "Role Title", "Exact Role Title"} + UNKNOWN_COMPANIES = {"", "Unknown Company", "Company", "Company Name", "Detected Company"} + UNKNOWN_LOCATIONS = {"", "Unknown", "Identified Location", "City, State or Remote/Hybrid", "Unknown Location"} - analysis_dict = analysis.model_dump() - analysis_dict["title"] = analysis.title - analysis_dict["company"] = analysis.company - analysis_dict["location"] = analysis.location + # Prefer scraped title/company/location if AI/heuristic returned generic placeholders + if (not analysis.title or analysis.title in UNKNOWN_TITLES) and scraped_title and scraped_title not in UNKNOWN_TITLES: + analysis.title = scraped_title + if (not analysis.company or analysis.company in UNKNOWN_COMPANIES) and scraped_company and scraped_company not in UNKNOWN_COMPANIES: + analysis.company = scraped_company + if (not analysis.location or analysis.location in UNKNOWN_LOCATIONS) and scraped_location and scraped_location not in UNKNOWN_LOCATIONS: + analysis.location = scraped_location - return { - "analysis": analysis_dict, - "raw_text": job_text, - "source_url": source_url, - "title": analysis.title, - "company": analysis.company, - "location": analysis.location, - "salary_range": analysis.salary_range, - "work_mode": analysis.work_mode, - "experience_level": analysis.experience_level, - "experience_required": analysis.experience_required, - "is_new_grad_role": analysis.is_new_grad_role, - "new_grad_criteria": analysis.new_grad_criteria, - "required_skills": analysis.required_skills, - "preferred_skills": analysis.preferred_skills, - "tech_stack": analysis.tech_stack, - "soft_skills": analysis.soft_skills, - "ats_keywords": analysis.ats_keywords, - "summary": analysis.summary, - } + analysis_dict = analysis.model_dump() + analysis_dict["title"] = analysis.title + analysis_dict["company"] = analysis.company + analysis_dict["location"] = analysis.location + + return { + "analysis": analysis_dict, + "raw_text": job_text, + "source_url": source_url, + "title": analysis.title, + "company": analysis.company, + "location": analysis.location, + "salary_range": analysis.salary_range, + "work_mode": analysis.work_mode, + "experience_level": analysis.experience_level, + "experience_required": analysis.experience_required, + "is_new_grad_role": analysis.is_new_grad_role, + "new_grad_criteria": analysis.new_grad_criteria, + "required_skills": analysis.required_skills, + "preferred_skills": analysis.preferred_skills, + "tech_stack": analysis.tech_stack, + "soft_skills": analysis.soft_skills, + "ats_keywords": analysis.ats_keywords, + "summary": analysis.summary, + } + finally: + lease_service.release(lease_id) from backend.services.document_parser import extract_text_from_file diff --git a/backend/routers/auth.py b/backend/routers/auth.py index 2d6fad6..14e801f 100644 --- a/backend/routers/auth.py +++ b/backend/routers/auth.py @@ -428,11 +428,23 @@ def confirm_password_reset(req: PasswordResetConfirm): @router.post("/google", response_model=AuthResponse) -def google_auth(req: GoogleAuthRequest): +def google_auth(req: GoogleAuthRequest, request: Request): token = req.credential.strip() if not token: raise HTTPException(status_code=400, detail="Google credential token is missing.") + storage = get_storage() + cfg = load_config() + client_ip = get_client_ip(request, trust_proxy_headers=getattr(cfg, "trust_proxy_headers", False)) + limiter = RateLimiter(storage) + allowed, _, retry_after = limiter.check("google_login", client_ip, limit=10, window_seconds=60) + if not allowed: + raise HTTPException( + status_code=429, + detail="Too many Google login attempts. Please try again later.", + headers={"Retry-After": str(int(retry_after))}, + ) + try: token_data = verify_google_id_token(token) except Exception: diff --git a/backend/services/rate_limiter.py b/backend/services/rate_limiter.py index ba98a27..792d196 100644 --- a/backend/services/rate_limiter.py +++ b/backend/services/rate_limiter.py @@ -1,5 +1,6 @@ import hashlib import random +import threading import time from typing import Optional, Tuple from fastapi import Request @@ -8,6 +9,7 @@ class RateLimiter: """Database-backed atomic rate limiter with sliding window reset.""" + _lock = threading.Lock() def __init__(self, storage: StorageService): self.storage = storage @@ -34,50 +36,82 @@ def check( self.cleanup_expired(now) except Exception: pass - with self.storage._get_cursor() as cursor: - cursor.execute( - self.storage._format_sql( - "SELECT count, reset_at FROM rate_limits WHERE bucket = ? AND subject = ?" - ), - (bucket, subject), - ) - row = cursor.fetchone() - if not row: - reset_at = now + window_seconds - cursor.execute( - self.storage._format_sql( - "INSERT INTO rate_limits (bucket, subject, count, reset_at) VALUES (?, ?, ?, ?)" - ), - (bucket, subject, 1, reset_at), - ) - return True, max(0, limit - 1), 0.0 - - r = dict(row) - count = int(r["count"]) - reset_at = float(r["reset_at"]) - - if now >= reset_at: - new_reset_at = now + window_seconds - cursor.execute( - self.storage._format_sql( - "UPDATE rate_limits SET count = 1, reset_at = ? WHERE bucket = ? AND subject = ?" - ), - (new_reset_at, bucket, subject), - ) - return True, max(0, limit - 1), 0.0 - - if count < limit: - new_count = count + 1 - cursor.execute( - self.storage._format_sql( - "UPDATE rate_limits SET count = ?, reset_at = ? WHERE bucket = ? AND subject = ?" - ), - (new_count, reset_at, bucket, subject), - ) - return True, max(0, limit - new_count), 0.0 + + with self._lock: + if self.storage.is_postgres: + with self.storage._get_cursor() as cursor: + cursor.execute( + """ + INSERT INTO rate_limits (bucket, subject, count, reset_at) + VALUES (%s, %s, 0, %s) + ON CONFLICT (bucket, subject) DO NOTHING + """, + (bucket, subject, now + window_seconds), + ) + cursor.execute( + "SELECT count, reset_at FROM rate_limits WHERE bucket = %s AND subject = %s FOR UPDATE", + (bucket, subject), + ) + row = cursor.fetchone() + r = dict(row) + count = int(r["count"]) + reset_at = float(r["reset_at"]) + + if count == 0 or now >= reset_at: + new_reset_at = now + window_seconds + cursor.execute( + "UPDATE rate_limits SET count = 1, reset_at = %s WHERE bucket = %s AND subject = %s", + (new_reset_at, bucket, subject), + ) + return True, max(0, limit - 1), 0.0 + + if count < limit: + new_count = count + 1 + cursor.execute( + "UPDATE rate_limits SET count = %s WHERE bucket = %s AND subject = %s", + (new_count, bucket, subject), + ) + return True, max(0, limit - new_count), 0.0 + else: + retry_after = max(1.0, reset_at - now) + return False, 0, retry_after else: - retry_after = max(1.0, reset_at - now) - return False, 0, retry_after + with self.storage._get_immediate_cursor() as cursor: + cursor.execute( + "SELECT count, reset_at FROM rate_limits WHERE bucket = ? AND subject = ?", + (bucket, subject), + ) + row = cursor.fetchone() + if not row: + reset_at = now + window_seconds + cursor.execute( + "INSERT INTO rate_limits (bucket, subject, count, reset_at) VALUES (?, ?, 1, ?)", + (bucket, subject, reset_at), + ) + return True, max(0, limit - 1), 0.0 + + r = dict(row) + count = int(r["count"]) + reset_at = float(r["reset_at"]) + + if now >= reset_at: + new_reset_at = now + window_seconds + cursor.execute( + "UPDATE rate_limits SET count = 1, reset_at = ? WHERE bucket = ? AND subject = ?", + (new_reset_at, bucket, subject), + ) + return True, max(0, limit - 1), 0.0 + + if count < limit: + new_count = count + 1 + cursor.execute( + "UPDATE rate_limits SET count = ?, reset_at = ? WHERE bucket = ? AND subject = ?", + (new_count, reset_at, bucket, subject), + ) + return True, max(0, limit - new_count), 0.0 + else: + retry_after = max(1.0, reset_at - now) + return False, 0, retry_after def cleanup_expired(self, now_epoch: Optional[float] = None) -> int: """Removes expired rate limit records.""" diff --git a/backend/services/resource_leases.py b/backend/services/resource_leases.py new file mode 100644 index 0000000..9ff66a9 --- /dev/null +++ b/backend/services/resource_leases.py @@ -0,0 +1,90 @@ +import time +import uuid +from typing import Optional +from fastapi import HTTPException +from backend.storage import StorageService + + +class ResourceLeases: + """Manages expiring in-flight operation leases across workers to bound concurrency.""" + + def __init__(self, storage: StorageService): + self.storage = storage + + def acquire( + self, + user_id: str, + kind: str, + ttl_seconds: int = 30, + max_user_leases: int = 2, + max_global_leases: int = 8, + now_epoch: Optional[float] = None, + ) -> str: + """Acquires a resource lease. Raises HTTPException(429) if capacity is exhausted.""" + now = now_epoch if now_epoch is not None else time.time() + expires_at = now + ttl_seconds + + # Clean expired leases before checking + self.cleanup_expired(now) + + with self.storage._get_immediate_cursor() as cursor: + # Check user active leases + cursor.execute( + self.storage._format_sql( + "SELECT COUNT(*) as count FROM resource_leases WHERE user_id = ? AND expires_at > ?" + ), + (user_id, now), + ) + row = cursor.fetchone() + user_count = int(row["count"]) if row else 0 + if user_count >= max_user_leases: + raise HTTPException( + status_code=429, + detail=f"Too many concurrent {kind} operations for this user. Please wait for in-flight tasks to complete.", + headers={"Retry-After": "5"}, + ) + + # Check global active leases + cursor.execute( + self.storage._format_sql( + "SELECT COUNT(*) as count FROM resource_leases WHERE kind = ? AND expires_at > ?" + ), + (kind, now), + ) + row = cursor.fetchone() + global_count = int(row["count"]) if row else 0 + if global_count >= max_global_leases: + raise HTTPException( + status_code=429, + detail=f"Global concurrency capacity reached for {kind}. Please retry shortly.", + headers={"Retry-After": "5"}, + ) + + lease_id = str(uuid.uuid4()) + cursor.execute( + self.storage._format_sql( + "INSERT INTO resource_leases (lease_id, user_id, kind, expires_at) VALUES (?, ?, ?, ?)" + ), + (lease_id, user_id, kind, expires_at), + ) + return lease_id + + def release(self, lease_id: str) -> None: + """Releases an active lease.""" + if not lease_id: + return + with self.storage._get_immediate_cursor() as cursor: + cursor.execute( + self.storage._format_sql("DELETE FROM resource_leases WHERE lease_id = ?"), + (lease_id,), + ) + + def cleanup_expired(self, now_epoch: Optional[float] = None) -> int: + """Purges expired leases from storage.""" + now = now_epoch if now_epoch is not None else time.time() + with self.storage._get_immediate_cursor() as cursor: + cursor.execute( + self.storage._format_sql("DELETE FROM resource_leases WHERE expires_at <= ?"), + (now,), + ) + return cursor.rowcount if hasattr(cursor, "rowcount") else 0 diff --git a/backend/storage.py b/backend/storage.py index 64b9896..5094aa1 100644 --- a/backend/storage.py +++ b/backend/storage.py @@ -161,6 +161,30 @@ def _get_cursor(self): finally: conn.close() + @contextmanager + def _get_immediate_cursor(self): + """Yields a cursor inside an immediate transaction for SQLite or normal cursor for PostgreSQL.""" + if self.is_postgres: + with self._get_cursor() as cursor: + yield cursor + else: + conn = sqlite3.connect(self.db_path, timeout=30.0) + conn.row_factory = sqlite3.Row + conn.isolation_level = None + cursor = conn.cursor() + try: + cursor.execute("BEGIN IMMEDIATE") + yield cursor + cursor.execute("COMMIT") + except Exception: + try: + cursor.execute("ROLLBACK") + except Exception: + pass + raise + finally: + conn.close() + def close(self): """Closes all connections in the pool if active.""" if self.is_postgres and self.pool: @@ -335,6 +359,17 @@ def _init_db(self): PRIMARY KEY (bucket, subject) ) """) + # Resource leases table + cursor.execute(""" + CREATE TABLE IF NOT EXISTS resource_leases ( + lease_id TEXT PRIMARY KEY, + user_id TEXT NOT NULL, + kind TEXT NOT NULL, + expires_at REAL NOT NULL + ) + """) + cursor.execute("CREATE INDEX IF NOT EXISTS idx_resource_leases_user ON resource_leases(user_id, kind)") + cursor.execute("CREATE INDEX IF NOT EXISTS idx_resource_leases_expires ON resource_leases(expires_at)") # --- Users CRUD --- def create_user( diff --git a/tests/test_atomic_limits.py b/tests/test_atomic_limits.py new file mode 100644 index 0000000..b3d4a7a --- /dev/null +++ b/tests/test_atomic_limits.py @@ -0,0 +1,50 @@ +from concurrent.futures import ThreadPoolExecutor +from threading import Barrier +from unittest.mock import Mock + +import pytest + +from backend import main +from backend.services.rate_limiter import RateLimiter + + +@pytest.mark.parametrize('seed', [0, 1]) +def test_concurrent_requests_obey_exact_budget(seed): + limiter = RateLimiter(main.storage) + if seed: + limiter.check('race', 'user', 3, 60) + start = Barrier(12) + def check(_): + start.wait() + return limiter.check('race', 'user', 3, 60)[0] + with ThreadPoolExecutor(max_workers=12) as workers: + assert sum(workers.map(check, range(12))) == 3 - seed + + +def test_authenticated_analysis_exhaustion_never_calls_scraper(client, monkeypatch): + user = main.storage.create_user('budget@example.test', None, 'Budget', email_verified=True) + main.app.dependency_overrides[main.get_optional_user] = lambda: user + scrape = Mock() + monkeypatch.setattr("backend.services.scraper.ScraperService.scrape_url", scrape) + try: + for _ in range(30): + RateLimiter(main.storage).check('user_ai_ops', user.id, 30, 60) + response = client.post('/api/analyze/job', json={'url': 'https://example.test/job'}) + assert response.status_code == 429 + assert int(response.headers['retry-after']) >= 1 + scrape.assert_not_called() + finally: + main.app.dependency_overrides.pop(main.get_optional_user, None) + + +def test_google_exhaustion_never_calls_verifier(client, monkeypatch): + from backend.routers import auth + verify = Mock(side_effect=ValueError('synthetic invalid token')) + monkeypatch.setattr(auth, 'verify_google_id_token', verify) + for _ in range(10): + assert client.post('/api/auth/google', json={'credential': 'synthetic'}).status_code == 401 + verify.reset_mock() + response = client.post('/api/auth/google', json={'credential': 'synthetic'}) + assert response.status_code == 429 + assert int(response.headers['retry-after']) >= 1 + verify.assert_not_called() diff --git a/tests/test_resource_leases.py b/tests/test_resource_leases.py new file mode 100644 index 0000000..673f70e --- /dev/null +++ b/tests/test_resource_leases.py @@ -0,0 +1,72 @@ +import pytest +from fastapi import HTTPException +from backend import main +from backend.services.resource_leases import ResourceLeases + + +def test_resource_lease_lifecycle(): + leases = ResourceLeases(main.storage) + lease_id = leases.acquire(user_id="user_1", kind="test_op", ttl_seconds=60) + assert lease_id is not None + leases.release(lease_id) + + # Releasing an empty or already released lease is safe and idempotent + leases.release(lease_id) + leases.release("") + + +def test_user_concurrency_limit(): + leases = ResourceLeases(main.storage) + lease1 = leases.acquire(user_id="user_concurrent", kind="test_op", ttl_seconds=60, max_user_leases=2) + lease2 = leases.acquire(user_id="user_concurrent", kind="test_op", ttl_seconds=60, max_user_leases=2) + + with pytest.raises(HTTPException) as exc_info: + leases.acquire(user_id="user_concurrent", kind="test_op", ttl_seconds=60, max_user_leases=2) + + assert exc_info.value.status_code == 429 + assert "Too many concurrent" in exc_info.value.detail + assert exc_info.value.headers.get("Retry-After") == "5" + + leases.release(lease1) + leases.release(lease2) + + +def test_global_concurrency_limit(): + leases = ResourceLeases(main.storage) + active = [] + try: + for i in range(5): + lid = leases.acquire(user_id=f"user_{i}", kind="global_op", ttl_seconds=60, max_global_leases=5) + active.append(lid) + + with pytest.raises(HTTPException) as exc_info: + leases.acquire(user_id="another_user", kind="global_op", ttl_seconds=60, max_global_leases=5) + + assert exc_info.value.status_code == 429 + assert "Global concurrency capacity reached" in exc_info.value.detail + finally: + for lid in active: + leases.release(lid) + + +def test_expired_leases_are_cleaned_up(): + leases = ResourceLeases(main.storage) + # Acquire with past timestamp (expired) + lid = leases.acquire( + user_id="user_expired", + kind="test_op", + ttl_seconds=10, + now_epoch=1000.0, + max_user_leases=1, + ) + + # At t=1020.0, the lease is expired. New acquisition at t=1020.0 must succeed. + lid2 = leases.acquire( + user_id="user_expired", + kind="test_op", + ttl_seconds=10, + now_epoch=1020.0, + max_user_leases=1, + ) + assert lid2 is not None + leases.release(lid2) From a0a2f93679280a467014f1649b448b2e3ebb7518 Mon Sep 17 00:00:00 2001 From: Sourav Chandhok <114119117+souravC01@users.noreply.github.com> Date: Fri, 11 Sep 2026 23:53:37 -0400 Subject: [PATCH 05/14] fix(resume): Enforce atomic quotas, idempotent delete, and Unicode downloads --- backend/main.py | 111 ++++++++++++----------------- backend/services/object_storage.py | 14 ++-- backend/services/resume_files.py | 94 ++++++++++++++++++++++++ backend/storage.py | 32 +++++++-- tests/conftest.py | 10 ++- tests/test_resume_lifecycle.py | 90 +++++++++++++++++++++++ 6 files changed, 271 insertions(+), 80 deletions(-) create mode 100644 backend/services/resume_files.py create mode 100644 tests/test_resume_lifecycle.py diff --git a/backend/main.py b/backend/main.py index 51982a0..e577696 100644 --- a/backend/main.py +++ b/backend/main.py @@ -39,7 +39,9 @@ from backend.services.ai_engine import AIEngine, extract_raw_content_from_response from backend.services.excel_exporter import ExcelExporter from backend.services.object_storage import ObjectStorageService +from backend.services.resume_files import ResumeFileManager, format_content_disposition from backend.services.outbound_http import SSRFBlockedError + from backend.routers.auth import router as auth_router, get_current_user, get_optional_user, set_storage_service @@ -171,6 +173,10 @@ async def add_security_headers(request: Request, call_next): else: print("[INFO] Cloudflare R2 not configured. Using local file storage fallback.") +def get_resume_file_manager() -> ResumeFileManager: + return ResumeFileManager(storage, object_storage) + + def get_ai_engine(user_id: Optional[str] = None) -> AIEngine: if not user_id: @@ -371,13 +377,16 @@ def add_resume(req: ResumeCreate, current_user: User = Depends(get_current_user) status_code=422, detail="Resume content exceeds maximum limit of 100,000 characters.", ) - user_resume_count = storage.count_user_resumes(current_user.id) - if user_resume_count >= MAX_RESUMES_PER_USER: - raise HTTPException( - status_code=422, - detail=f"Maximum limit of {MAX_RESUMES_PER_USER} resumes reached per account.", + try: + return storage.add_resume( + name=req.name, + content=req.content, + file_key=None, + user_id=current_user.id, + max_resumes=MAX_RESUMES_PER_USER, ) - return storage.add_resume(name=req.name, content=req.content, file_key=None, user_id=current_user.id) + except ValueError as ve: + raise HTTPException(status_code=422, detail=str(ve)) @app.post("/api/resumes/upload", response_model=Resume) @@ -409,13 +418,6 @@ def upload_resume_file( detail="File exceeds maximum allowed size of 10MB.", ) - current_bytes = storage.get_user_upload_bytes(current_user.id) - if current_bytes + len(content_bytes) > MAX_USER_STORAGE_BYTES: - raise HTTPException( - status_code=422, - detail="Maximum total storage limit of 50MB reached for resumes.", - ) - if content_override and content_override.strip(): if len(content_override) > 100_000: raise HTTPException( @@ -429,40 +431,26 @@ def upload_resume_file( raise HTTPException(status_code=400, detail="No readable text could be extracted from this document.") final_content = extracted_text.strip() - # Upload binary to Cloudflare R2 or local directory scoped to user_id + resume_name = name.strip() if (name and name.strip()) else Path(file.filename).stem try: - file_key = object_storage.upload_file( - content_bytes=content_bytes, + resume_file_manager = get_resume_file_manager() + resume = resume_file_manager.reserve_and_upload( + + user_id=current_user.id, filename=file.filename, content_type=file.content_type, - user_id=current_user.id, + content_bytes=content_bytes, + resume_name=resume_name, + text_content=final_content, + max_resumes=MAX_RESUMES_PER_USER, + max_storage_bytes=MAX_USER_STORAGE_BYTES, ) - except RuntimeError as re: + except ValueError as ve: + raise HTTPException(status_code=422, detail=str(ve)) + except RuntimeError: raise HTTPException(status_code=503, detail="Storage service temporarily unavailable.") - - storage_backend = "r2" if object_storage.is_configured else "local" - attachment = storage.create_attachment( - user_id=current_user.id, - storage_backend=storage_backend, - object_key=file_key, - original_filename=file.filename, - content_type=file.content_type, - size_bytes=len(content_bytes), - ) - - resume_name = name.strip() if (name and name.strip()) else Path(file.filename).stem - try: - resume = storage.add_resume( - name=resume_name, - content=final_content, - file_key=file_key, - user_id=current_user.id, - attachment_id=attachment.id, - ) except Exception as e: - object_storage.delete_file(file_key, user_id=current_user.id) - storage.update_attachment_deletion_state(attachment.id, "failed", user_id=current_user.id) - raise HTTPException(status_code=500, detail="Failed to save resume record.") + raise HTTPException(status_code=500, detail=f"Failed to save resume record: {str(e)}") resume.download_url = f"/api/resumes/{resume.id}/download" return resume @@ -474,6 +462,7 @@ def upload_resume_file( raise HTTPException(status_code=500, detail=f"Failed to process file: {str(e)}") + @app.post("/api/resumes/parse-file") def parse_resume_file( file: UploadFile = File(...), @@ -571,7 +560,7 @@ def download_resume_file( content=file_bytes, media_type=content_type, headers={ - "Content-Disposition": f'attachment; filename="{safe_filename}"', + "Content-Disposition": format_content_disposition(raw_filename), "Access-Control-Expose-Headers": "Content-Disposition, X-Fallback-Generated", }, ) @@ -581,12 +570,13 @@ def download_resume_file( if raw_text: fallback_bytes = generate_docx_from_text(title=resume.name or "Resume", text=raw_text) safe_name = re.sub(r'[^a-zA-Z0-9_-]', '_', resume.name or "resume").strip('_') or "resume" - fallback_filename = f"{safe_name}_generated.docx" + fallback_raw = f"{safe_name}_generated.docx" return Response( + content=fallback_bytes, media_type="application/vnd.openxmlformats-officedocument.wordprocessingml.document", headers={ - "Content-Disposition": f'attachment; filename="{fallback_filename}"', + "Content-Disposition": format_content_disposition(fallback_raw), "X-Fallback-Generated": "true", "Access-Control-Expose-Headers": "Content-Disposition, X-Fallback-Generated", }, @@ -616,29 +606,18 @@ def update_resume( @app.delete("/api/resumes/{resume_id}") def delete_resume(resume_id: str, current_user: User = Depends(get_current_user)): - resume = storage.get_resume(resume_id, user_id=current_user.id) - if not resume: - raise HTTPException(status_code=404, detail="Resume not found.") + try: + resume_file_manager = get_resume_file_manager() + deleted = resume_file_manager.delete_resume(user_id=current_user.id, resume_id=resume_id) + + if not deleted: + raise HTTPException(status_code=404, detail="Resume not found.") + return {"success": True} + except HTTPException: + raise + except Exception as e: + raise HTTPException(status_code=500, detail=str(e)) - if resume.attachment_id or resume.file_key: - attachment = None - if resume.attachment_id: - attachment = storage.get_attachment(resume.attachment_id, user_id=current_user.id) - elif resume.file_key: - attachment = storage.get_attachment_by_key(resume.file_key, user_id=current_user.id) - - if attachment: - storage.update_attachment_deletion_state(attachment.id, "pending", user_id=current_user.id) - deleted = object_storage.delete_file(attachment.object_key, user_id=current_user.id) - if not deleted: - storage.update_attachment_deletion_state(attachment.id, "failed", user_id=current_user.id) - raise HTTPException(status_code=500, detail="Failed to delete associated storage file.") - storage.delete_attachment(attachment.id, user_id=current_user.id) - elif resume.file_key: - object_storage.delete_file(resume.file_key, user_id=current_user.id) - - storage.delete_resume(resume_id, user_id=current_user.id) - return {"success": True} class MatchResumesRequest(BaseModel): diff --git a/backend/services/object_storage.py b/backend/services/object_storage.py index fbc0a73..a73af2d 100644 --- a/backend/services/object_storage.py +++ b/backend/services/object_storage.py @@ -203,11 +203,15 @@ def delete_file(self, object_key: str, user_id: Optional[str] = None) -> bool: return False safe_path = self._resolve_safe_path(object_key, user_id=user_id) - if safe_path and safe_path.is_file(): - try: - safe_path.unlink() + if safe_path: + if not safe_path.exists(): return True - except Exception: - return False + if safe_path.is_file(): + try: + safe_path.unlink() + return True + except Exception: + return False return False + diff --git a/backend/services/resume_files.py b/backend/services/resume_files.py new file mode 100644 index 0000000..8f0a87a --- /dev/null +++ b/backend/services/resume_files.py @@ -0,0 +1,94 @@ +import re +from typing import Optional +from urllib.parse import quote + +from backend.storage import StorageService +from backend.services.object_storage import ObjectStorageService +from backend.models import Resume + + +def format_content_disposition(filename: str) -> str: + """Builds RFC 6266 and RFC 5987 compliant Content-Disposition header with UTF-8 filename*.""" + clean_ascii = re.sub(r'[^\x20-\x7E]', '_', filename) + clean_ascii = re.sub(r'[\r\n"\\]', '_', clean_ascii).strip() or "resume.pdf" + encoded_utf8 = quote(filename.replace('\r', '_').replace('\n', '_'), safe="") + return f'attachment; filename="{clean_ascii}"; filename*=UTF-8\'\'{encoded_utf8}' + + +class ResumeFileManager: + def __init__(self, storage: StorageService, object_storage: ObjectStorageService): + self.storage = storage + self.object_storage = object_storage + + def reserve_and_upload( + self, + user_id: str, + filename: str, + content_type: Optional[str], + content_bytes: bytes, + resume_name: Optional[str], + text_content: str, + max_resumes: int = 10, + max_storage_bytes: int = 50 * 1024 * 1024, + ) -> Resume: + current_bytes = self.storage.get_user_upload_bytes(user_id) + if current_bytes + len(content_bytes) > max_storage_bytes: + raise ValueError("Maximum total storage limit of 50MB reached for resumes.") + + file_key = self.object_storage.upload_file( + content_bytes=content_bytes, + filename=filename, + content_type=content_type, + user_id=user_id, + ) + + storage_backend = "r2" if self.object_storage.is_configured else "local" + attachment = self.storage.create_attachment( + user_id=user_id, + storage_backend=storage_backend, + object_key=file_key, + original_filename=filename, + content_type=content_type, + size_bytes=len(content_bytes), + ) + + final_name = resume_name.strip() if (resume_name and resume_name.strip()) else filename + try: + resume = self.storage.add_resume( + name=final_name, + content=text_content, + file_key=file_key, + user_id=user_id, + attachment_id=attachment.id, + max_resumes=max_resumes, + ) + return resume + except Exception: + self.object_storage.delete_file(file_key, user_id=user_id) + self.storage.update_attachment_deletion_state(attachment.id, "failed", user_id=user_id) + raise + + def delete_resume(self, user_id: str, resume_id: str) -> bool: + resume = self.storage.get_resume(resume_id, user_id=user_id) + if not resume: + return False + + if resume.attachment_id or resume.file_key: + attachment = None + if resume.attachment_id: + attachment = self.storage.get_attachment(resume.attachment_id, user_id=user_id) + elif resume.file_key: + attachment = self.storage.get_attachment_by_key(resume.file_key, user_id=user_id) + + if attachment: + self.storage.update_attachment_deletion_state(attachment.id, "pending", user_id=user_id) + deleted = self.object_storage.delete_file(attachment.object_key, user_id=user_id) + if not deleted: + self.storage.update_attachment_deletion_state(attachment.id, "failed", user_id=user_id) + raise RuntimeError("Failed to delete associated storage file.") + self.storage.delete_attachment(attachment.id, user_id=user_id) + elif resume.file_key: + self.object_storage.delete_file(resume.file_key, user_id=user_id) + + self.storage.delete_resume(resume_id, user_id=user_id) + return True diff --git a/backend/storage.py b/backend/storage.py index 5094aa1..9ebafe6 100644 --- a/backend/storage.py +++ b/backend/storage.py @@ -1,6 +1,7 @@ import json import os import sqlite3 +import threading import uuid from contextlib import contextmanager from datetime import datetime, timezone, timedelta @@ -64,7 +65,10 @@ def canonical_posting_url(url: Optional[str]) -> Optional[str]: return clean.rstrip("/") + class StorageService: + _resume_quota_lock = threading.Lock() + def __init__(self, db_path: str = "data/tracker.db", force_sqlite: bool = False, database_url: Optional[str] = None): if force_sqlite: raw_url = None @@ -562,16 +566,30 @@ def add_resume( file_key: Optional[str] = None, user_id: Optional[str] = None, attachment_id: Optional[str] = None, + max_resumes: Optional[int] = None, ) -> Resume: now = datetime.now().isoformat() resume_id = str(uuid.uuid4()) - with self._get_cursor() as cursor: - cursor.execute( - self._format_sql( - "INSERT INTO resumes (id, user_id, name, content, file_key, attachment_id, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)" - ), - (resume_id, user_id, name, content, file_key, attachment_id, now, now), - ) + with self._resume_quota_lock: + with self._get_immediate_cursor() as cursor: + if user_id and max_resumes is not None: + if self.is_postgres: + cursor.execute("SELECT id FROM users WHERE id = %s FOR UPDATE", (user_id,)) + cursor.execute( + self._format_sql("SELECT COUNT(*) as cnt FROM resumes WHERE user_id = ?"), + (user_id,), + ) + row = cursor.fetchone() + count = int(row["cnt"]) if isinstance(row, dict) else int(row[0]) + if count >= max_resumes: + raise ValueError(f"Maximum limit of {max_resumes} resumes reached per account.") + + cursor.execute( + self._format_sql( + "INSERT INTO resumes (id, user_id, name, content, file_key, attachment_id, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)" + ), + (resume_id, user_id, name, content, file_key, attachment_id, now, now), + ) return Resume( id=resume_id, name=name, diff --git a/tests/conftest.py b/tests/conftest.py index b881f72..3d9ed02 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -70,10 +70,16 @@ def get_file(self, object_key, user_id=None): def delete_file(self, object_key, user_id=None): target = self.root / object_key + if not target.exists(): + return True if not target.is_file(): return False - target.unlink() - return True + try: + target.unlink() + return True + except Exception: + return False + @pytest.fixture(autouse=True) diff --git a/tests/test_resume_lifecycle.py b/tests/test_resume_lifecycle.py new file mode 100644 index 0000000..c83a399 --- /dev/null +++ b/tests/test_resume_lifecycle.py @@ -0,0 +1,90 @@ +import io +from concurrent.futures import ThreadPoolExecutor +from threading import Barrier +import pytest +from backend import main + + +def test_concurrent_resume_creation_obeys_quota_limit(two_users): + client, alpha, _ = two_users + + # Fill up to 9 resumes + for i in range(9): + resp = client.post("/api/resumes", headers=alpha, json={ + "name": f"Resume {i}", + "content": f"Software Engineer Experience {i}", + }) + assert resp.status_code == 200 + + # User now has 9 resumes. Only 1 slot remains before hitting the 10-resume cap. + start = Barrier(4) + def attempt_create(idx): + start.wait() + return client.post("/api/resumes", headers=alpha, json={ + "name": f"Concurrent Resume {idx}", + "content": f"Concurrent Software Engineer Experience {idx}", + }) + + with ThreadPoolExecutor(max_workers=4) as workers: + results = list(workers.map(attempt_create, range(4))) + + successes = [r for r in results if r.status_code == 200] + rejected = [r for r in results if r.status_code == 422] + + assert len(successes) == 1 + assert len(rejected) == 3 + + # Confirm database contains exactly 10 resumes for this user + all_resumes = client.get("/api/resumes", headers=alpha).json() + assert len(all_resumes) == 10 + + +def test_unicode_resume_filename_download(two_users): + client, alpha, _ = two_users + + unicode_filename = "简历_Sourav_2026.txt" + file_content = b"Resume content in UTF-8: Engineering lead with NLP experience." + + upload_resp = client.post( + "/api/resumes/upload", + headers=alpha, + files={"file": (unicode_filename, io.BytesIO(file_content), "text/plain")}, + ) + assert upload_resp.status_code == 200 + resume_id = upload_resp.json()["id"] + + download_resp = client.get(f"/api/resumes/{resume_id}/download", headers=alpha) + assert download_resp.status_code == 200 + assert download_resp.content == file_content + + + # Content-Disposition header must be ASCII-safe and include filename* + disposition = download_resp.headers.get("content-disposition", "") + assert "filename*=" in disposition + assert "UTF-8''" in disposition + + +def test_delete_resume_with_missing_binary_is_idempotent(two_users): + client, alpha, _ = two_users + + file_content = b"Resume text with binary that will disappear." + upload_resp = client.post( + "/api/resumes/upload", + headers=alpha, + files={"file": ("to_delete.txt", io.BytesIO(file_content), "text/plain")}, + ) + assert upload_resp.status_code == 200 + resume_id = upload_resp.json()["id"] + + # Delete underlying storage file so it's missing + resume = main.storage.get_resume(resume_id) + if resume and resume.file_key: + main.object_storage.delete_file(resume.file_key) + + # Deleting the resume when binary is already missing must succeed idempotently (200), not 500 + del_resp = client.delete(f"/api/resumes/{resume_id}", headers=alpha) + assert del_resp.status_code == 200 + + # Second delete returns 404 because resume is gone + del_resp2 = client.delete(f"/api/resumes/{resume_id}", headers=alpha) + assert del_resp2.status_code == 404 From 7f9e32988bcf3cef770adbec81e1df0dae110b1d Mon Sep 17 00:00:00 2001 From: Sourav Chandhok <114119117+souravC01@users.noreply.github.com> Date: Fri, 11 Sep 2026 23:59:51 -0400 Subject: [PATCH 06/14] fix(auth): Add Brevo email delivery, account recovery landing page, and SPA fallback --- backend/config.py | 19 ++ backend/main.py | 31 ++- backend/routers/auth.py | 10 +- backend/services/email_service.py | 51 +++++ frontend/src/App.jsx | 35 ++++ frontend/src/api/client.js | 30 +++ frontend/src/components/AuthLinkPage.jsx | 239 +++++++++++++++++++++++ frontend/src/test/AuthLinkPage.test.jsx | 80 ++++++++ tests/test_email_service.py | 78 ++++++++ tests/test_spa_fallback.py | 45 +++++ 10 files changed, 615 insertions(+), 3 deletions(-) create mode 100644 frontend/src/components/AuthLinkPage.jsx create mode 100644 frontend/src/test/AuthLinkPage.test.jsx create mode 100644 tests/test_email_service.py create mode 100644 tests/test_spa_fallback.py diff --git a/backend/config.py b/backend/config.py index 77fd16d..f1fd0ca 100644 --- a/backend/config.py +++ b/backend/config.py @@ -56,6 +56,10 @@ class AppConfig: smtp_user: Optional[str] = None smtp_password: Optional[str] = None smtp_from: Optional[str] = None + brevo_api_key: Optional[str] = None + brevo_sender_email: Optional[str] = None + brevo_sender_name: Optional[str] = "JobHelperGuru" + def _get_or_create_local_secrets(secrets_dir: Optional[str] = None) -> dict: @@ -117,6 +121,11 @@ def load_config() -> AppConfig: smtp_password = os.getenv("SMTP_PASSWORD") smtp_from = os.getenv("SMTP_FROM") or os.getenv("SMTP_USER") + # Brevo configuration + brevo_api_key = os.getenv("BREVO_API_KEY") + brevo_sender_email = os.getenv("BREVO_SENDER_EMAIL") + brevo_sender_name = os.getenv("BREVO_SENDER_NAME") or "JobHelperGuru" + if raw_mode == "production": jwt_key = (os.getenv("JWT_SECRET_KEY") or "").strip() if not jwt_key: @@ -149,6 +158,9 @@ def load_config() -> AppConfig: smtp_user=smtp_user, smtp_password=smtp_password, smtp_from=smtp_from, + brevo_api_key=brevo_api_key, + brevo_sender_email=brevo_sender_email, + brevo_sender_name=brevo_sender_name, ) elif raw_mode == "local": @@ -171,6 +183,9 @@ def load_config() -> AppConfig: smtp_user=smtp_user, smtp_password=smtp_password, smtp_from=smtp_from, + brevo_api_key=brevo_api_key, + brevo_sender_email=brevo_sender_email, + brevo_sender_name=brevo_sender_name, ) else: # test mode @@ -192,5 +207,9 @@ def load_config() -> AppConfig: smtp_user=smtp_user, smtp_password=smtp_password, smtp_from=smtp_from, + brevo_api_key=brevo_api_key, + brevo_sender_email=brevo_sender_email, + brevo_sender_name=brevo_sender_name, ) + diff --git a/backend/main.py b/backend/main.py index e577696..99f0d1e 100644 --- a/backend/main.py +++ b/backend/main.py @@ -4,11 +4,14 @@ from typing import Optional, List from fastapi import FastAPI, HTTPException, Response, Depends, UploadFile, File, Form, Request -from fastapi.responses import RedirectResponse +from fastapi.responses import RedirectResponse, FileResponse from fastapi.middleware.cors import CORSMiddleware + from fastapi.staticfiles import StaticFiles from pydantic import BaseModel from starlette.types import ASGIApp, Receive, Scope, Send +from starlette.exceptions import HTTPException as StarletteHTTPException + from backend.config import load_config from backend.services.rate_limiter import RateLimiter, get_client_ip @@ -861,7 +864,31 @@ def test_ai(req: SettingsUpdate, current_user: User = Depends(get_current_user)) } +class SPAStaticFiles(StaticFiles): + """ + StaticFiles extension that serves index.html for Single Page Application + deep links (e.g. /verify-email, /reset-password) when the requested static + path does not exist and is not an API call. + """ + + async def get_response(self, path: str, scope): + try: + return await super().get_response(path, scope) + except (HTTPException, StarletteHTTPException) as ex: + normalized_path = path.replace("\\", "/").lstrip("/") + if ex.status_code == 404 and not normalized_path.startswith("api/") and normalized_path != "api": + index_path = Path(self.directory) / "index.html" + if index_path.is_file(): + return FileResponse(str(index_path)) + raise + + + + + + # --- Static frontend files mounting --- dist_path = (Path(__file__).resolve().parent.parent / "frontend" / "dist").resolve() if dist_path.exists() and dist_path.is_dir(): - app.mount("/", StaticFiles(directory=str(dist_path), html=True), name="frontend") + app.mount("/", SPAStaticFiles(directory=str(dist_path), html=True), name="frontend") + diff --git a/backend/routers/auth.py b/backend/routers/auth.py index 14e801f..e473391 100644 --- a/backend/routers/auth.py +++ b/backend/routers/auth.py @@ -30,7 +30,7 @@ hash_password, verify_password, ) -from backend.services.email_service import EmailService, SMTPEmailTransport +from backend.services.email_service import EmailService, SMTPEmailTransport, BrevoEmailTransport router = APIRouter(prefix="/api/auth", tags=["auth"]) oauth2_scheme = HTTPBearer(auto_error=False) @@ -38,6 +38,13 @@ def _create_email_service() -> EmailService: cfg = load_config() + if cfg.brevo_api_key: + transport = BrevoEmailTransport( + api_key=cfg.brevo_api_key, + sender_email=cfg.brevo_sender_email or "noreply@jobhelper.guru", + sender_name=cfg.brevo_sender_name or "JobHelperGuru", + ) + return EmailService(transport=transport) if cfg.smtp_host: transport = SMTPEmailTransport( host=cfg.smtp_host, @@ -50,6 +57,7 @@ def _create_email_service() -> EmailService: return EmailService() + # Email service instance (mockable in tests) email_service = _create_email_service() diff --git a/backend/services/email_service.py b/backend/services/email_service.py index 832fd83..9c63ff3 100644 --- a/backend/services/email_service.py +++ b/backend/services/email_service.py @@ -77,7 +77,58 @@ def send(self, recipient: str, subject: str, text_body: str) -> None: server.send_message(msg) +class BrevoEmailTransport(EmailTransport): + """Brevo (formerly Sendinblue) v3 API email delivery transport.""" + + def __init__( + self, + api_key: str, + sender_email: str = "noreply@jobhelper.guru", + sender_name: str = "JobHelperGuru", + endpoint: str = "https://api.brevo.com/v3/smtp/email", + timeout: float = 10.0, + ): + self.api_key = api_key + self.sender_email = sender_email + self.sender_name = sender_name + self.endpoint = endpoint + self.timeout = timeout + + def send(self, recipient: str, subject: str, text_body: str) -> None: + import json + import urllib.request + import urllib.error + + payload = { + "sender": {"name": self.sender_name, "email": self.sender_email}, + "to": [{"email": recipient}], + "subject": subject, + "textContent": text_body, + } + data = json.dumps(payload).encode("utf-8") + req = urllib.request.Request( + self.endpoint, + data=data, + headers={ + "api-key": self.api_key, + "Content-Type": "application/json", + "Accept": "application/json", + }, + method="POST", + ) + try: + with urllib.request.urlopen(req, timeout=self.timeout) as resp: + if resp.status not in (200, 201, 202): + raise RuntimeError(f"Brevo API returned unexpected status {resp.status}") + except urllib.error.HTTPError as e: + err_msg = e.read().decode("utf-8", errors="replace") + raise RuntimeError(f"Brevo API error ({e.code}): {err_msg}") + except Exception as e: + raise RuntimeError(f"Failed to deliver email via Brevo: {e}") + + class EmailService: + """Service for dispatching transactional authentication emails.""" def __init__(self, transport: Optional[EmailTransport] = None): diff --git a/frontend/src/App.jsx b/frontend/src/App.jsx index 02fb9df..1b88786 100644 --- a/frontend/src/App.jsx +++ b/frontend/src/App.jsx @@ -20,6 +20,7 @@ import SettingsModal from './components/SettingsModal'; import OfflineSwitchModal from './components/OfflineSwitchModal'; import AuthModal from './components/AuthModal'; import PrivacyModal from './components/PrivacyModal'; +import AuthLinkPage from './components/AuthLinkPage'; import UserNav from './components/UserNav'; import ThemeToggle from './components/ThemeToggle'; import { @@ -36,6 +37,14 @@ export default function App() { const [currentUser, setCurrentUser] = useState(getCurrentUser()); const [isAuthOpen, setIsAuthOpen] = useState(false); const [authMode, setAuthMode] = useState('login'); + const [currentPath, setCurrentPath] = useState(() => (typeof window !== 'undefined' ? window.location.pathname : '/')); + + useEffect(() => { + const onPopState = () => setCurrentPath(window.location.pathname); + window.addEventListener('popstate', onPopState); + return () => window.removeEventListener('popstate', onPopState); + }, []); + const [theme, setTheme] = useState(() => { return localStorage.getItem('jobhelperguru_theme') || 'light'; @@ -219,7 +228,33 @@ export default function App() { const offerCount = applications.filter((a) => a.status === 'Offered').length; const totalCount = applications.length; + if (currentPath === '/verify-email' || currentPath === '/reset-password') { + return ( + <> + setCurrentPath('/')} + onOpenAuth={(mode) => { + setCurrentPath('/'); + setAuthMode(mode || 'login'); + setIsAuthOpen(true); + }} + /> + setIsAuthOpen(false)} + onSuccess={(user) => { + setCurrentUser(user); + setIsAuthOpen(false); + }} + initialMode={authMode} + /> + + ); + } + return ( +
{/* Top Corporate Navbar */}
diff --git a/frontend/src/api/client.js b/frontend/src/api/client.js index f47fe78..9c531e5 100644 --- a/frontend/src/api/client.js +++ b/frontend/src/api/client.js @@ -43,6 +43,13 @@ async function parseApiError(res, defaultMsg) { let errorMsg = defaultMsg; let canSwitchOffline = res.status === 502; let modelName = ''; + let retryAfter = null; + + const retryHeader = res.headers?.get?.('Retry-After'); + if (retryHeader) { + const parsed = parseInt(retryHeader, 10); + if (!isNaN(parsed)) retryAfter = parsed; + } try { const data = await res.json(); @@ -63,9 +70,23 @@ async function parseApiError(res, defaultMsg) { err.canSwitchOffline = canSwitchOffline; err.modelName = modelName; err.status = res.status; + if (retryAfter !== null) { + err.retryAfter = retryAfter; + } return err; } +export async function authFetchWithRetry(url, options = {}, maxRetries = 1) { + const res = await authFetch(url, options); + if (res.status === 429 && maxRetries > 0) { + const retryHeader = res.headers?.get?.('Retry-After'); + const delaySec = Math.min(Math.max(parseInt(retryHeader || '1', 10), 1), 10); + await new Promise((resolve) => setTimeout(resolve, delaySec * 1000)); + return authFetchWithRetry(url, options, maxRetries - 1); + } + return res; +} + // --- Auth APIs --- export async function registerUser({ email, password, name }) { @@ -106,6 +127,10 @@ export async function requestEmailVerification(email) { headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ email }), }); + if (!res.ok) { + const data = await res.json().catch(() => ({ detail: 'Failed to request verification email' })); + throw new Error(data.detail || 'Failed to request verification email'); + } return res.json(); } @@ -128,6 +153,10 @@ export async function requestPasswordReset(email) { headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ email }), }); + if (!res.ok) { + const data = await res.json().catch(() => ({ detail: 'Failed to request password reset' })); + throw new Error(data.detail || 'Failed to request password reset'); + } return res.json(); } @@ -144,6 +173,7 @@ export async function confirmPasswordReset(token, new_password) { return res.json(); } + export async function googleAuthUser(credential) { const res = await fetch(`${API_BASE}/auth/google`, { method: 'POST', diff --git a/frontend/src/components/AuthLinkPage.jsx b/frontend/src/components/AuthLinkPage.jsx new file mode 100644 index 0000000..607211b --- /dev/null +++ b/frontend/src/components/AuthLinkPage.jsx @@ -0,0 +1,239 @@ +import { useState, useEffect } from 'react'; +import { Mail, Lock, CheckCircle, AlertCircle, Loader2, ArrowLeft, Eye, EyeOff } from 'lucide-react'; +import { confirmEmailVerification, confirmPasswordReset } from '../api/client'; + +export default function AuthLinkPage({ path, onGoHome, onOpenAuth }) { + const isVerify = path === '/verify-email'; + const queryParams = new URLSearchParams(window.location.search); + const token = queryParams.get('token') || ''; + + // Verification state + const [verifyStatus, setVerifyStatus] = useState('loading'); // 'loading' | 'success' | 'error' + const [verifyError, setVerifyError] = useState(''); + + // Password reset state + const [password, setPassword] = useState(''); + const [confirmPassword, setConfirmPassword] = useState(''); + const [showPassword, setShowPassword] = useState(false); + const [resetLoading, setResetLoading] = useState(false); + const [resetError, setResetError] = useState(''); + const [resetSuccess, setResetSuccess] = useState(false); + + useEffect(() => { + if (isVerify) { + if (!token) { + setVerifyStatus('error'); + setVerifyError('Verification link is missing the token parameter.'); + return; + } + confirmEmailVerification(token) + .then(() => { + setVerifyStatus('success'); + }) + .catch((err) => { + setVerifyStatus('error'); + setVerifyError(err.message || 'Verification link is invalid or has expired.'); + }); + } + }, [isVerify, token]); + + const handleResetSubmit = async (e) => { + e.preventDefault(); + setResetError(''); + if (!token) { + setResetError('Reset token is missing.'); + return; + } + if (password.length < 8) { + setResetError('Password must be at least 8 characters long.'); + return; + } + if (password !== confirmPassword) { + setResetError('Passwords do not match.'); + return; + } + + setResetLoading(true); + try { + await confirmPasswordReset(token, password); + setResetSuccess(true); + } catch (err) { + setResetError(err.message || 'Failed to reset password. The link may have expired.'); + } finally { + setResetLoading(false); + } + }; + + const handleNavigateHome = () => { + window.history.pushState({}, '', '/'); + if (onGoHome) onGoHome(); + }; + + const handleOpenLogin = () => { + window.history.pushState({}, '', '/'); + if (onGoHome) onGoHome(); + if (onOpenAuth) onOpenAuth('login'); + }; + + return ( +
+
+
+ +
+ + {isVerify ? ( +
+ {verifyStatus === 'loading' && ( + <> +
+ +
+

Verifying your email...

+

+ Please hold on while we confirm your email address. +

+ + )} + + {verifyStatus === 'success' && ( + <> +
+
+ +
+
+

Email Verified!

+

+ Your email address has been verified successfully. You can now sign in to your account. +

+
+ +
+ + )} + + {verifyStatus === 'error' && ( + <> +
+
+ +
+
+

Verification Failed

+

+ {verifyError} +

+
+ +
+ + )} +
+ ) : ( +
+
+

Reset Your Password

+

+ Enter your new account password below. +

+
+ + {resetSuccess ? ( +
+
+
+ +
+
+

Password Updated!

+

+ Your password has been reset successfully. You can now sign in with your new credentials. +

+ +
+ ) : ( +
+ {resetError && ( +
+ + {resetError} +
+ )} + +
+ +
+ setPassword(e.target.value)} + required + placeholder="At least 8 characters" + className="w-full pl-10 pr-10 py-2.5 text-sm rounded-xl border border-slate-200 dark:border-slate-800 bg-white dark:bg-slate-950 text-slate-900 dark:text-slate-100 focus:outline-none focus:ring-2 focus:ring-primary-500" + /> + + +
+
+ +
+ +
+ setConfirmPassword(e.target.value)} + required + placeholder="Re-enter your new password" + className="w-full pl-10 pr-10 py-2.5 text-sm rounded-xl border border-slate-200 dark:border-slate-800 bg-white dark:bg-slate-950 text-slate-900 dark:text-slate-100 focus:outline-none focus:ring-2 focus:ring-primary-500" + /> + +
+
+ + + +
+ )} +
+ )} +
+
+ ); +} diff --git a/frontend/src/test/AuthLinkPage.test.jsx b/frontend/src/test/AuthLinkPage.test.jsx new file mode 100644 index 0000000..e2af66f --- /dev/null +++ b/frontend/src/test/AuthLinkPage.test.jsx @@ -0,0 +1,80 @@ +import { describe, expect, it, vi, beforeEach, afterEach } from 'vitest'; +import { render, screen, waitFor, fireEvent } from '@testing-library/react'; +import AuthLinkPage from '../components/AuthLinkPage'; +import * as client from '../api/client'; + +describe('AuthLinkPage Component', () => { + beforeEach(() => { + vi.restoreAllMocks(); + }); + + it('verifies email with valid token and displays success state', async () => { + delete window.location; + window.location = new URL('https://jobhelper.guru/verify-email?token=valid-token-123'); + + vi.spyOn(client, 'confirmEmailVerification').mockResolvedValue({ message: 'Verified' }); + + render(); + + expect(screen.getByText(/Verifying your email/i)).toBeInTheDocument(); + + await waitFor(() => { + expect(screen.getByText(/Email Verified!/i)).toBeInTheDocument(); + }); + expect(screen.getByRole('button', { name: /Sign In/i })).toBeInTheDocument(); + }); + + it('displays error when email verification fails', async () => { + delete window.location; + window.location = new URL('https://jobhelper.guru/verify-email?token=expired-token'); + + vi.spyOn(client, 'confirmEmailVerification').mockRejectedValue(new Error('Token expired')); + + render(); + + await waitFor(() => { + expect(screen.getByText(/Verification Failed/i)).toBeInTheDocument(); + expect(screen.getByText(/Token expired/i)).toBeInTheDocument(); + }); + }); + + it('renders password reset form and submits successfully', async () => { + delete window.location; + window.location = new URL('https://jobhelper.guru/reset-password?token=reset-token-456'); + + const resetSpy = vi.spyOn(client, 'confirmPasswordReset').mockResolvedValue({ message: 'Password reset' }); + + render(); + + expect(screen.getByText(/Reset Your Password/i)).toBeInTheDocument(); + + const newPassInput = screen.getByLabelText(/^New Password/i); + const confirmPassInput = screen.getByLabelText(/^Confirm New Password/i); + fireEvent.change(newPassInput, { target: { value: 'NewSecurePassword123!' } }); + fireEvent.change(confirmPassInput, { target: { value: 'NewSecurePassword123!' } }); + + fireEvent.click(screen.getByRole('button', { name: /Set New Password/i })); + + await waitFor(() => { + expect(resetSpy).toHaveBeenCalledWith('reset-token-456', 'NewSecurePassword123!'); + expect(screen.getByText(/Password Updated!/i)).toBeInTheDocument(); + }); + }); + + it('shows validation error when passwords do not match', async () => { + delete window.location; + window.location = new URL('https://jobhelper.guru/reset-password?token=reset-token-456'); + + render(); + + const newPassInput = screen.getByLabelText(/^New Password/i); + const confirmPassInput = screen.getByLabelText(/^Confirm New Password/i); + fireEvent.change(newPassInput, { target: { value: 'NewSecurePassword123!' } }); + fireEvent.change(confirmPassInput, { target: { value: 'DifferentPassword456!' } }); + + fireEvent.click(screen.getByRole('button', { name: /Set New Password/i })); + + + expect(screen.getByText(/Passwords do not match/i)).toBeInTheDocument(); + }); +}); diff --git a/tests/test_email_service.py b/tests/test_email_service.py new file mode 100644 index 0000000..41816dc --- /dev/null +++ b/tests/test_email_service.py @@ -0,0 +1,78 @@ +import json +from unittest.mock import patch, MagicMock +import pytest +from backend.services.email_service import ( + BrevoEmailTransport, + EmailService, + FakeEmailTransport, +) +from backend.config import AppConfig +from backend.routers import auth + + +def test_brevo_transport_sends_correct_payload(): + transport = BrevoEmailTransport( + api_key="test-brevo-key-12345", + sender_email="alerts@jobhelper.guru", + sender_name="JobHelperGuru Alerts", + ) + + mock_resp = MagicMock() + mock_resp.status = 201 + mock_resp.__enter__.return_value = mock_resp + + with patch("urllib.request.urlopen", return_value=mock_resp) as mock_urlopen: + transport.send( + recipient="candidate@example.com", + subject="Welcome to JobHelperGuru", + text_body="Hello candidate!", + ) + + assert mock_urlopen.called + req = mock_urlopen.call_args[0][0] + assert req.full_url == "https://api.brevo.com/v3/smtp/email" + assert req.headers["Api-key"] == "test-brevo-key-12345" + assert req.headers["Content-type"] == "application/json" + + body = json.loads(req.data.decode("utf-8")) + assert body["sender"]["email"] == "alerts@jobhelper.guru" + assert body["sender"]["name"] == "JobHelperGuru Alerts" + assert body["to"] == [{"email": "candidate@example.com"}] + assert body["subject"] == "Welcome to JobHelperGuru" + assert body["textContent"] == "Hello candidate!" + + +def test_brevo_transport_handles_http_errors(): + import urllib.error + transport = BrevoEmailTransport(api_key="invalid-key") + + mock_err = urllib.error.HTTPError( + url="https://api.brevo.com/v3/smtp/email", + code=401, + msg="Unauthorized", + hdrs={}, + fp=MagicMock(read=lambda: b'{"message": "Key not valid"}'), + ) + + with patch("urllib.request.urlopen", side_effect=mock_err): + with pytest.raises(RuntimeError, match="Brevo API error"): + transport.send( + recipient="user@example.com", + subject="Test", + text_body="Test", + ) + + +def test_create_email_service_selects_brevo_when_configured(monkeypatch): + mock_cfg = AppConfig( + brevo_api_key="brevo-api-key-xyz", + brevo_sender_email="sender@jobhelper.guru", + brevo_sender_name="My App", + ) + monkeypatch.setattr("backend.routers.auth.load_config", lambda: mock_cfg) + + service = auth._create_email_service() + assert isinstance(service.transport, BrevoEmailTransport) + assert service.transport.api_key == "brevo-api-key-xyz" + assert service.transport.sender_email == "sender@jobhelper.guru" + assert service.transport.sender_name == "My App" diff --git a/tests/test_spa_fallback.py b/tests/test_spa_fallback.py new file mode 100644 index 0000000..52e3518 --- /dev/null +++ b/tests/test_spa_fallback.py @@ -0,0 +1,45 @@ +import pytest +from starlette.testclient import TestClient +from backend.main import app, SPAStaticFiles +from pathlib import Path + + +def test_spa_fallback_serves_index_for_deep_links(tmp_path): + # Create fake dist directory with index.html and an asset + dist = tmp_path / "dist" + dist.mkdir() + index_file = dist / "index.html" + index_file.write_text("
SPA Root
", encoding="utf-8") + asset_file = dist / "style.css" + asset_file.write_text("body { color: red; }", encoding="utf-8") + + from fastapi import FastAPI + test_app = FastAPI() + + @test_app.get("/api/data") + def get_data(): + return {"ok": True} + + test_app.mount("/", SPAStaticFiles(directory=str(dist), html=True), name="frontend") + client = TestClient(test_app) + + # Direct asset returns asset + assert client.get("/style.css").status_code == 200 + assert client.get("/style.css").text == "body { color: red; }" + + # Root returns index.html + assert client.get("/").status_code == 200 + assert "SPA Root" in client.get("/").text + + # Deep links return index.html + resp_verify = client.get("/verify-email?token=abc123xyz") + assert resp_verify.status_code == 200 + assert "SPA Root" in resp_verify.text + + resp_reset = client.get("/reset-password?token=def456uvw") + assert resp_reset.status_code == 200 + assert "SPA Root" in resp_reset.text + + # API 404 does not return index.html + resp_api = client.get("/api/not-found") + assert resp_api.status_code == 404 From cfb59dc6b576817736b3f75a4388edf84e09ba2c Mon Sep 17 00:00:00 2001 From: Sourav Chandhok <114119117+souravC01@users.noreply.github.com> Date: Sat, 12 Sep 2026 00:03:11 -0400 Subject: [PATCH 07/14] fix(data): Add data repair script and enforce strict application validation --- backend/repair_data.py | 164 ++++++++++++++++++++++++++++++++++++++ tests/test_data_repair.py | 137 +++++++++++++++++++++++++++++++ 2 files changed, 301 insertions(+) create mode 100644 backend/repair_data.py create mode 100644 tests/test_data_repair.py diff --git a/backend/repair_data.py b/backend/repair_data.py new file mode 100644 index 0000000..6ffa877 --- /dev/null +++ b/backend/repair_data.py @@ -0,0 +1,164 @@ +import argparse +import os +import sys +from typing import Optional, Dict, Any + +from backend.storage import StorageService +from backend.services.object_storage import ObjectStorageService + + +def repair_database( + storage: StorageService, + object_storage: Optional[ObjectStorageService] = None, + dry_run: bool = True, +) -> Dict[str, Any]: + """ + Scans and repairs data anomalies: + 1. Applications with NULL or empty status -> 'Wishlist' + 2. Applications with NULL required_skills or ats_keywords -> '[]' + 3. User settings with invalid default_follow_up_days -> 7 + 4. Orphaned active attachments not referenced by any resume -> cleaned up + """ + report = { + "corrupt_applications": 0, + "corrupt_settings": 0, + "orphaned_attachments": 0, + "applied": not dry_run, + } + + # 1. Inspect applications + with storage._get_cursor() as cursor: + cursor.execute( + storage._format_sql( + """ + SELECT id, status, required_skills, ats_keywords + FROM applications + WHERE status IS NULL OR status = '' OR required_skills IS NULL OR ats_keywords IS NULL + """ + ) + ) + corrupt_apps = cursor.fetchall() + report["corrupt_applications"] = len(corrupt_apps) + + if not dry_run and corrupt_apps: + for app_row in corrupt_apps: + app_id = app_row["id"] if isinstance(app_row, dict) else app_row[0] + cursor.execute( + storage._format_sql( + """ + UPDATE applications + SET status = COALESCE(NULLIF(status, ''), 'Wishlist'), + required_skills = COALESCE(required_skills, '[]'), + ats_keywords = COALESCE(ats_keywords, '[]') + WHERE id = ? + """ + ), + (app_id,), + ) + + # 2. Inspect user settings + with storage._get_cursor() as cursor: + cursor.execute( + storage._format_sql( + """ + SELECT user_id, key, value + FROM user_settings + WHERE key = 'default_follow_up_days' + """ + ) + ) + settings_rows = cursor.fetchall() + corrupt_user_settings = [] + for s_row in settings_rows: + uid = s_row["user_id"] if isinstance(s_row, dict) else s_row[0] + val = s_row["value"] if isinstance(s_row, dict) else s_row[2] + try: + days = int(val) + if days < 1 or days > 90: + corrupt_user_settings.append(uid) + except (ValueError, TypeError): + corrupt_user_settings.append(uid) + + # Global settings table + cursor.execute("SELECT key, value FROM settings WHERE key = 'default_follow_up_days'") + global_row = cursor.fetchone() + global_corrupt = False + if global_row: + gval = global_row["value"] if isinstance(global_row, dict) else global_row[1] + try: + gdays = int(gval) + if gdays < 1 or gdays > 90: + global_corrupt = True + except (ValueError, TypeError): + global_corrupt = True + + report["corrupt_settings"] = len(corrupt_user_settings) + (1 if global_corrupt else 0) + + if not dry_run: + for uid in corrupt_user_settings: + cursor.execute( + storage._format_sql( + "UPDATE user_settings SET value = '7' WHERE user_id = ? AND key = 'default_follow_up_days'" + ), + (uid,), + ) + if global_corrupt: + cursor.execute("UPDATE settings SET value = '7' WHERE key = 'default_follow_up_days'") + + # 3. Inspect orphaned attachments + with storage._get_cursor() as cursor: + cursor.execute( + storage._format_sql( + """ + SELECT id, user_id, object_key + FROM attachments + WHERE deletion_state = 'active' + AND id NOT IN (SELECT attachment_id FROM resumes WHERE attachment_id IS NOT NULL) + """ + ) + ) + orphans = cursor.fetchall() + report["orphaned_attachments"] = len(orphans) + + if not dry_run and orphans: + for att_row in orphans: + att_id = att_row["id"] if isinstance(att_row, dict) else att_row[0] + user_id = att_row["user_id"] if isinstance(att_row, dict) else att_row[1] + object_key = att_row["object_key"] if isinstance(att_row, dict) else att_row[2] + + if object_storage and object_key: + try: + object_storage.delete_file(object_key, user_id=user_id) + except Exception: + pass + + cursor.execute( + storage._format_sql( + "UPDATE attachments SET deletion_state = 'orphaned_cleaned' WHERE id = ?" + ), + (att_id,), + ) + + return report + + +def main_cli(): + parser = argparse.ArgumentParser(description="Repair database inconsistencies and clean orphaned attachments.") + parser.add_argument("--apply", action="store_true", help="Apply fixes directly to the database (default is dry-run).") + parser.add_argument("--db-path", default=None, help="Optional path to local SQLite database.") + args = parser.parse_args() + + db_path = args.db_path or os.getenv("JOB_HELPER_DB", "data/tracker.db") + storage = StorageService(db_path=db_path) + object_storage = ObjectStorageService() + + dry_run = not args.apply + print(f"[{'DRY RUN' if dry_run else 'APPLYING REPAIRS'}] Starting database audit...") + result = repair_database(storage, object_storage, dry_run=dry_run) + print("Audit results:", result) + if dry_run: + print("Run with --apply to commit changes to the database.") + + +if __name__ == "__main__": + main_cli() diff --git a/tests/test_data_repair.py b/tests/test_data_repair.py new file mode 100644 index 0000000..039cce7 --- /dev/null +++ b/tests/test_data_repair.py @@ -0,0 +1,137 @@ +import pytest +from backend import main +from backend.models import ApplicationCreate +from backend.repair_data import repair_database + + +def test_reject_null_status_in_application_update(two_users): + client, alpha, _ = two_users + + create_resp = client.post("/api/applications", headers=alpha, json={ + "company": "Test Co", + "role": "Engineer", + "status": "Wishlist", + }) + assert create_resp.status_code == 200 + app_id = create_resp.json()["id"] + + # Explicit null status must be rejected with 422 + patch_resp = client.patch(f"/api/applications/{app_id}", headers=alpha, json={ + "status": None, + }) + assert patch_resp.status_code == 422 + + +def test_repair_database_fixes_null_status_and_orphans(two_users): + client, alpha, _ = two_users + storage = main.storage + + # Insert raw corrupted records directly into database bypassing validation + with storage._get_cursor() as cursor: + cursor.execute( + storage._format_sql( + """ + INSERT INTO applications ( + id, user_id, company, role, status, location, salary, url, + required_skills, ats_keywords, date_added, application_date, + follow_up_date, notes, best_resume_id, created_at, updated_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """ + ), + ( + "corrupt-app-1", + "user-1", + "Legacy Corp", + "Developer", + "", # Empty status + "Remote", + "$100k", + "https://example.com", + None, # Null skills + None, # Null keywords + "2026-01-01", + "", + "", + "", + None, + "2026-01-01T00:00:00", + "2026-01-01T00:00:00", + ), + ) + + # Create an orphaned attachment + cursor.execute( + storage._format_sql( + """ + INSERT INTO attachments ( + id, user_id, storage_backend, object_key, original_filename, + content_type, size_bytes, deletion_state, created_at, updated_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """ + ), + ( + "orphan-att-1", + "user-1", + "local", + "resumes/user-1/orphan.pdf", + "orphan.pdf", + "application/pdf", + 1024, + "active", + "2026-01-01T00:00:00", + "2026-01-01T00:00:00", + ), + ) + + # Create invalid default_follow_up_days in user_settings + cursor.execute( + storage._format_sql( + "INSERT INTO user_settings (user_id, key, value) VALUES (?, ?, ?)" + ), + ("user-1", "default_follow_up_days", "999"), + ) + + # 1. Dry run should report findings without mutating + dry_report = repair_database(storage=storage, object_storage=main.object_storage, dry_run=True) + assert dry_report["corrupt_applications"] == 1 + assert dry_report["corrupt_settings"] == 1 + assert dry_report["orphaned_attachments"] == 1 + assert dry_report["applied"] is False + + # Check database was not modified yet + with storage._get_cursor() as cursor: + cursor.execute(storage._format_sql("SELECT status FROM applications WHERE id = ?"), ("corrupt-app-1",)) + row = cursor.fetchone() + val = row["status"] if isinstance(row, dict) else row[0] + assert val == "" + + cursor.execute(storage._format_sql("SELECT value FROM user_settings WHERE user_id = ? AND key = ?"), ("user-1", "default_follow_up_days")) + row = cursor.fetchone() + s_val = row["value"] if isinstance(row, dict) else row[0] + assert s_val == "999" + + # 2. Apply repair + apply_report = repair_database(storage=storage, object_storage=main.object_storage, dry_run=False) + assert apply_report["corrupt_applications"] == 1 + assert apply_report["corrupt_settings"] == 1 + assert apply_report["orphaned_attachments"] == 1 + assert apply_report["applied"] is True + + # Check database was fixed + with storage._get_cursor() as cursor: + cursor.execute(storage._format_sql("SELECT status, required_skills FROM applications WHERE id = ?"), ("corrupt-app-1",)) + row = cursor.fetchone() + status_val = row["status"] if isinstance(row, dict) else row[0] + skills_val = row["required_skills"] if isinstance(row, dict) else row[1] + assert status_val == "Wishlist" + assert skills_val == "[]" + + cursor.execute(storage._format_sql("SELECT value FROM user_settings WHERE user_id = ? AND key = ?"), ("user-1", "default_follow_up_days")) + row = cursor.fetchone() + s_val = row["value"] if isinstance(row, dict) else row[0] + assert s_val == "7" + + cursor.execute(storage._format_sql("SELECT deletion_state FROM attachments WHERE id = ?"), ("orphan-att-1",)) + att_row = cursor.fetchone() + del_state = att_row["deletion_state"] if isinstance(att_row, dict) else att_row[0] + assert del_state in ("deleted", "orphaned_cleaned") From d9241d6067d1ea5978967974a2d963262e9cbb3a Mon Sep 17 00:00:00 2001 From: Sourav Chandhok <114119117+souravC01@users.noreply.github.com> Date: Sat, 12 Sep 2026 00:04:43 -0400 Subject: [PATCH 08/14] fix(migrate): Support PostgreSQL targets, key-value settings, and ledger tracking --- backend/migrate.py | 458 ++++++++++++++++++++++------------------ tests/test_migration.py | 89 +++++++- 2 files changed, 336 insertions(+), 211 deletions(-) diff --git a/backend/migrate.py b/backend/migrate.py index 6949fb0..0a20d14 100644 --- a/backend/migrate.py +++ b/backend/migrate.py @@ -71,6 +71,14 @@ def get_unowned_count(table: str) -> int: "user_settings_count": user_settings_count, "global_settings_count": global_settings_count, "provider_profiles_count": provider_profiles_count, + "migrated_users": 0, + "migrated_applications": 0, + "migrated_resumes": 0, + "migrated_attachments": 0, + "migrated_user_settings": 0, + "migrated_global_settings": 0, + "migrated_provider_profiles": 0, + "errors": [], } if dry_run: @@ -78,219 +86,245 @@ def get_unowned_count(table: str) -> int: return summary # Apply migration to destination - dest_storage = StorageService(db_path=dest_target) - now = datetime.now().isoformat() + is_postgres = dest_target.startswith("postgres://") or dest_target.startswith("postgresql://") + if is_postgres: + dest_storage = StorageService(database_url=dest_target) + else: + dest_storage = StorageService(db_path=dest_target, force_sqlite=True) - with dest_storage._get_cursor() as dst_cursor: - # Create migration ledger - dst_cursor.execute(""" - CREATE TABLE IF NOT EXISTS _migration_ledger ( - source_id TEXT, - table_name TEXT, - target_id TEXT, - migrated_at TEXT, - PRIMARY KEY (table_name, source_id) - ) - """) + now = datetime.now().isoformat() - # 1. Migrate Users - try: - src_users = src_cursor.execute("SELECT * FROM users").fetchall() - for u in src_users: - # Check ledger - dst_cursor.execute( - dest_storage._format_sql("SELECT 1 FROM _migration_ledger WHERE table_name = ? AND source_id = ?"), - ("users", u["id"]), - ) - if dst_cursor.fetchone(): - continue - pwd = u["hashed_password"] if "hashed_password" in u.keys() else (u["password_hash"] if "password_hash" in u.keys() else None) - ev = bool(u["email_verified"]) if "email_verified" in u.keys() else False - sv = int(u["session_version"]) if "session_version" in u.keys() else 1 - gsub = u["google_sub"] if "google_sub" in u.keys() else None - # Insert user - dst_cursor.execute( - dest_storage._format_sql( - "INSERT INTO users (id, email, hashed_password, name, avatar_url, provider, email_verified, session_version, google_sub, created_at, updated_at) " - "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) " - + ("ON CONFLICT (id) DO NOTHING" if dest_storage.is_postgres else "ON CONFLICT(id) DO NOTHING") - ), - (u["id"], u["email"], pwd, u["name"], u["avatar_url"], u["provider"], ev, sv, gsub, u["created_at"], u["updated_at"]), - ) - dst_cursor.execute( - dest_storage._format_sql("INSERT INTO _migration_ledger VALUES (?, ?, ?, ?)"), - (u["id"], "users", u["id"], now), + try: + with dest_storage._get_cursor() as dst_cursor: + # Create migration ledger + dst_cursor.execute(""" + CREATE TABLE IF NOT EXISTS _migration_ledger ( + source_id TEXT, + table_name TEXT, + target_id TEXT, + migrated_at TEXT, + PRIMARY KEY (table_name, source_id) ) - except Exception as e: - print(f"[WARN] Error migrating users: {e}") + """) - # 2. Migrate Attachments - try: - src_attachments = src_cursor.execute("SELECT * FROM attachments").fetchall() - for a in src_attachments: - dst_cursor.execute( - dest_storage._format_sql("SELECT 1 FROM _migration_ledger WHERE table_name = ? AND source_id = ?"), - ("attachments", a["id"]), - ) - if dst_cursor.fetchone(): - continue - target_user_id = a["user_id"] or map_unowned_to - dst_cursor.execute( - dest_storage._format_sql( - "INSERT INTO attachments (id, user_id, storage_backend, object_key, original_filename, content_type, size_bytes, deletion_state, created_at, updated_at) " - "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) " - + ("ON CONFLICT (id) DO NOTHING" if dest_storage.is_postgres else "ON CONFLICT(id) DO NOTHING") - ), - (a["id"], target_user_id, a["storage_backend"], a["object_key"], a["original_filename"], a["content_type"], a["size_bytes"], a["deletion_state"], a["created_at"], a["updated_at"]), - ) - dst_cursor.execute( - dest_storage._format_sql("INSERT INTO _migration_ledger VALUES (?, ?, ?, ?)"), - (a["id"], "attachments", a["id"], now), - ) - except Exception as e: - print(f"[WARN] Error migrating attachments: {e}") + # 1. Migrate Users + try: + src_users = src_cursor.execute("SELECT * FROM users").fetchall() + for u in src_users: + # Check ledger + dst_cursor.execute( + dest_storage._format_sql("SELECT 1 FROM _migration_ledger WHERE table_name = ? AND source_id = ?"), + ("users", u["id"]), + ) + if dst_cursor.fetchone(): + continue + pwd = u["hashed_password"] if "hashed_password" in u.keys() else (u["password_hash"] if "password_hash" in u.keys() else None) + ev = bool(u["email_verified"]) if "email_verified" in u.keys() else False + sv = int(u["session_version"]) if "session_version" in u.keys() else 1 + gsub = u["google_sub"] if "google_sub" in u.keys() else None + # Insert user + dst_cursor.execute( + dest_storage._format_sql( + "INSERT INTO users (id, email, hashed_password, name, avatar_url, provider, email_verified, session_version, google_sub, created_at, updated_at) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) " + + ("ON CONFLICT (id) DO NOTHING" if dest_storage.is_postgres else "ON CONFLICT(id) DO NOTHING") + ), + (u["id"], u["email"], pwd, u["name"], u["avatar_url"], u["provider"], ev, sv, gsub, u["created_at"], u["updated_at"]), + ) + dst_cursor.execute( + dest_storage._format_sql("INSERT INTO _migration_ledger VALUES (?, ?, ?, ?)"), + (u["id"], "users", u["id"], now), + ) + summary["migrated_users"] += 1 + except Exception as e: + summary["errors"].append(f"users: {e}") + print(f"[WARN] Error migrating users: {e}") - # 3. Migrate Resumes - try: - src_resumes = src_cursor.execute("SELECT * FROM resumes").fetchall() - for r in src_resumes: - dst_cursor.execute( - dest_storage._format_sql("SELECT 1 FROM _migration_ledger WHERE table_name = ? AND source_id = ?"), - ("resumes", r["id"]), - ) - if dst_cursor.fetchone(): - continue - target_user_id = r["user_id"] or map_unowned_to - att_id = r["attachment_id"] if "attachment_id" in r.keys() else None - dst_cursor.execute( - dest_storage._format_sql( - "INSERT INTO resumes (id, user_id, name, content, file_key, attachment_id, created_at, updated_at) " - "VALUES (?, ?, ?, ?, ?, ?, ?, ?) " - + ("ON CONFLICT (id) DO NOTHING" if dest_storage.is_postgres else "ON CONFLICT(id) DO NOTHING") - ), - (r["id"], target_user_id, r["name"], r["content"], r["file_key"], att_id, r["created_at"], r["updated_at"]), - ) - dst_cursor.execute( - dest_storage._format_sql("INSERT INTO _migration_ledger VALUES (?, ?, ?, ?)"), - (r["id"], "resumes", r["id"], now), - ) - except Exception as e: - print(f"[WARN] Error migrating resumes: {e}") + # 2. Migrate Attachments + try: + src_attachments = src_cursor.execute("SELECT * FROM attachments").fetchall() + for a in src_attachments: + dst_cursor.execute( + dest_storage._format_sql("SELECT 1 FROM _migration_ledger WHERE table_name = ? AND source_id = ?"), + ("attachments", a["id"]), + ) + if dst_cursor.fetchone(): + continue + target_user_id = a["user_id"] or map_unowned_to + dst_cursor.execute( + dest_storage._format_sql( + "INSERT INTO attachments (id, user_id, storage_backend, object_key, original_filename, content_type, size_bytes, deletion_state, created_at, updated_at) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) " + + ("ON CONFLICT (id) DO NOTHING" if dest_storage.is_postgres else "ON CONFLICT(id) DO NOTHING") + ), + (a["id"], target_user_id, a["storage_backend"], a["object_key"], a["original_filename"], a["content_type"], a["size_bytes"], a["deletion_state"], a["created_at"], a["updated_at"]), + ) + dst_cursor.execute( + dest_storage._format_sql("INSERT INTO _migration_ledger VALUES (?, ?, ?, ?)"), + (a["id"], "attachments", a["id"], now), + ) + summary["migrated_attachments"] += 1 + except Exception as e: + summary["errors"].append(f"attachments: {e}") + print(f"[WARN] Error migrating attachments: {e}") - # 4. Migrate Applications - try: - src_apps = src_cursor.execute("SELECT * FROM applications").fetchall() - for app in src_apps: - dst_cursor.execute( - dest_storage._format_sql("SELECT 1 FROM _migration_ledger WHERE table_name = ? AND source_id = ?"), - ("applications", app["id"]), - ) - if dst_cursor.fetchone(): - continue - target_user_id = app["user_id"] or map_unowned_to - date_added = app["date_added"] if "date_added" in app.keys() else app["created_at"] - app_date = app["application_date"] if "application_date" in app.keys() else None - dst_cursor.execute( - dest_storage._format_sql( - "INSERT INTO applications (id, user_id, company, role, status, location, salary, url, required_skills, ats_keywords, date_added, application_date, follow_up_date, notes, best_resume_id, created_at, updated_at) " - "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) " - + ("ON CONFLICT (id) DO NOTHING" if dest_storage.is_postgres else "ON CONFLICT(id) DO NOTHING") - ), - ( - app["id"], - target_user_id, - app["company"], - app["role"], - app["status"], - app["location"], - app["salary"], - app["url"], - app["required_skills"], - app["ats_keywords"], - date_added, - app_date, - app["follow_up_date"], - app["notes"], - app["best_resume_id"], - app["created_at"], - app["updated_at"], - ), - ) - dst_cursor.execute( - dest_storage._format_sql("INSERT INTO _migration_ledger VALUES (?, ?, ?, ?)"), - (app["id"], "applications", app["id"], now), - ) - except Exception as e: - print(f"[WARN] Error migrating applications: {e}") + # 3. Migrate Resumes + try: + src_resumes = src_cursor.execute("SELECT * FROM resumes").fetchall() + for r in src_resumes: + dst_cursor.execute( + dest_storage._format_sql("SELECT 1 FROM _migration_ledger WHERE table_name = ? AND source_id = ?"), + ("resumes", r["id"]), + ) + if dst_cursor.fetchone(): + continue + target_user_id = r["user_id"] or map_unowned_to + att_id = r["attachment_id"] if "attachment_id" in r.keys() else None + dst_cursor.execute( + dest_storage._format_sql( + "INSERT INTO resumes (id, user_id, name, content, file_key, attachment_id, created_at, updated_at) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?) " + + ("ON CONFLICT (id) DO NOTHING" if dest_storage.is_postgres else "ON CONFLICT(id) DO NOTHING") + ), + (r["id"], target_user_id, r["name"], r["content"], r["file_key"], att_id, r["created_at"], r["updated_at"]), + ) + dst_cursor.execute( + dest_storage._format_sql("INSERT INTO _migration_ledger VALUES (?, ?, ?, ?)"), + (r["id"], "resumes", r["id"], now), + ) + summary["migrated_resumes"] += 1 + except Exception as e: + summary["errors"].append(f"resumes: {e}") + print(f"[WARN] Error migrating resumes: {e}") - # 5. Migrate User Settings - try: - src_user_settings = src_cursor.execute("SELECT * FROM user_settings").fetchall() - for s in src_user_settings: - dst_cursor.execute( - dest_storage._format_sql( - "INSERT INTO user_settings (user_id, key, value, created_at, updated_at) " - "VALUES (?, ?, ?, ?, ?) " - + ("ON CONFLICT (user_id, key) DO NOTHING" if dest_storage.is_postgres else "ON CONFLICT(user_id, key) DO NOTHING") - ), - (s["user_id"], s["key"], s["value"], s["created_at"], s["updated_at"]), - ) - except Exception: - pass + # 4. Migrate Applications + try: + src_apps = src_cursor.execute("SELECT * FROM applications").fetchall() + for app in src_apps: + dst_cursor.execute( + dest_storage._format_sql("SELECT 1 FROM _migration_ledger WHERE table_name = ? AND source_id = ?"), + ("applications", app["id"]), + ) + if dst_cursor.fetchone(): + continue + target_user_id = app["user_id"] or map_unowned_to + date_added = app["date_added"] if "date_added" in app.keys() else app["created_at"] + app_date = app["application_date"] if "application_date" in app.keys() else None + dst_cursor.execute( + dest_storage._format_sql( + "INSERT INTO applications (id, user_id, company, role, status, location, salary, url, required_skills, ats_keywords, date_added, application_date, follow_up_date, notes, best_resume_id, created_at, updated_at) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) " + + ("ON CONFLICT (id) DO NOTHING" if dest_storage.is_postgres else "ON CONFLICT(id) DO NOTHING") + ), + ( + app["id"], + target_user_id, + app["company"], + app["role"], + app["status"], + app["location"], + app["salary"], + app["url"], + app["required_skills"], + app["ats_keywords"], + date_added, + app_date, + app["follow_up_date"], + app["notes"], + app["best_resume_id"], + app["created_at"], + app["updated_at"], + ), + ) + dst_cursor.execute( + dest_storage._format_sql("INSERT INTO _migration_ledger VALUES (?, ?, ?, ?)"), + (app["id"], "applications", app["id"], now), + ) + summary["migrated_applications"] += 1 + except Exception as e: + summary["errors"].append(f"applications: {e}") + print(f"[WARN] Error migrating applications: {e}") - # 6. Migrate Global Settings - try: - src_settings = src_cursor.execute("SELECT * FROM settings").fetchall() - for gs in src_settings: - dst_cursor.execute( - dest_storage._format_sql( - "INSERT INTO settings (key, value) " - "VALUES (?, ?) " - + ("ON CONFLICT (key) DO NOTHING" if dest_storage.is_postgres else "ON CONFLICT(key) DO NOTHING") - ), - (gs["key"], gs["value"]), - ) - except Exception: - pass + # 5. Migrate User Settings + try: + src_user_settings = src_cursor.execute("SELECT * FROM user_settings").fetchall() + for s in src_user_settings: + dst_cursor.execute( + dest_storage._format_sql( + "INSERT INTO user_settings (user_id, key, value) " + "VALUES (?, ?, ?) " + + ("ON CONFLICT (user_id, key) DO UPDATE SET value = excluded.value" if dest_storage.is_postgres else "ON CONFLICT(user_id, key) DO UPDATE SET value = excluded.value") + ), + (s["user_id"], s["key"], s["value"]), + ) + summary["migrated_user_settings"] += 1 + except Exception as e: + summary["errors"].append(f"user_settings: {e}") + print(f"[WARN] Error migrating user_settings: {e}") - # 7. Migrate Provider Profiles - try: - src_profiles = src_cursor.execute("SELECT * FROM provider_profiles").fetchall() - for p in src_profiles: - dst_cursor.execute( - dest_storage._format_sql("SELECT 1 FROM _migration_ledger WHERE table_name = ? AND source_id = ?"), - ("provider_profiles", p["id"]), - ) - if dst_cursor.fetchone(): - continue - target_user_id = p["user_id"] or map_unowned_to - dst_cursor.execute( - dest_storage._format_sql( - "INSERT INTO provider_profiles (id, user_id, name, api_base_url, model_name, api_key_encrypted, key_suffix, is_active, created_at, updated_at) " - "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) " - + ("ON CONFLICT (id) DO NOTHING" if dest_storage.is_postgres else "ON CONFLICT(id) DO NOTHING") - ), - ( - p["id"], - target_user_id, - p["name"], - p["api_base_url"], - p["model_name"], - p["api_key_encrypted"], - p["key_suffix"], - p["is_active"], - p["created_at"], - p["updated_at"], - ), - ) - dst_cursor.execute( - dest_storage._format_sql("INSERT INTO _migration_ledger VALUES (?, ?, ?, ?)"), - (p["id"], "provider_profiles", p["id"], now), - ) - except Exception: - pass + # 6. Migrate Global Settings + try: + src_settings = src_cursor.execute("SELECT * FROM settings").fetchall() + for gs in src_settings: + dst_cursor.execute( + dest_storage._format_sql( + "INSERT INTO settings (key, value) " + "VALUES (?, ?) " + + ("ON CONFLICT (key) DO UPDATE SET value = excluded.value" if dest_storage.is_postgres else "ON CONFLICT(key) DO UPDATE SET value = excluded.value") + ), + (gs["key"], gs["value"]), + ) + summary["migrated_global_settings"] += 1 + except Exception as e: + summary["errors"].append(f"settings: {e}") + print(f"[WARN] Error migrating settings: {e}") + + # 7. Migrate Provider Profiles + try: + src_profiles = src_cursor.execute("SELECT * FROM provider_profiles").fetchall() + for p in src_profiles: + dst_cursor.execute( + dest_storage._format_sql("SELECT 1 FROM _migration_ledger WHERE table_name = ? AND source_id = ?"), + ("provider_profiles", p["id"]), + ) + if dst_cursor.fetchone(): + continue + target_user_id = p["user_id"] or map_unowned_to + dst_cursor.execute( + dest_storage._format_sql( + "INSERT INTO provider_profiles (id, user_id, name, api_base_url, model_name, api_key_encrypted, key_suffix, is_active, created_at, updated_at) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) " + + ("ON CONFLICT (id) DO NOTHING" if dest_storage.is_postgres else "ON CONFLICT(id) DO NOTHING") + ), + ( + p["id"], + target_user_id, + p["name"], + p["api_base_url"], + p["model_name"], + p["api_key_encrypted"], + p["key_suffix"], + p["is_active"], + p["created_at"], + p["updated_at"], + ), + ) + dst_cursor.execute( + dest_storage._format_sql("INSERT INTO _migration_ledger VALUES (?, ?, ?, ?)"), + (p["id"], "provider_profiles", p["id"], now), + ) + summary["migrated_provider_profiles"] += 1 + except Exception as e: + summary["errors"].append(f"provider_profiles: {e}") + print(f"[WARN] Error migrating provider_profiles: {e}") + + finally: + dest_storage.close() + src_conn.close() + + if summary["errors"]: + summary["success"] = False - src_conn.close() return summary @@ -316,13 +350,17 @@ def main(): print(f"Mode: {'DRY RUN' if res['dry_run'] else 'APPLIED'}") print(f"Source: {res['source']}") print(f"Destination: {res['destination']}") - print(f"Users: {res['users_count']}") - print(f"Applications: {res['applications_count']} (unowned: {res['unowned_applications']})") - print(f"Resumes: {res['resumes_count']} (unowned: {res['unowned_resumes']})") - print(f"Attachments: {res['attachments_count']}") - print(f"User Settings: {res['user_settings_count']}") - print(f"Global Settings: {res['global_settings_count']}") - print(f"Provider Profiles: {res['provider_profiles_count']}") + print(f"Users: {res['users_count']} (migrated: {res.get('migrated_users', 0)})") + print(f"Applications: {res['applications_count']} (migrated: {res.get('migrated_applications', 0)}, unowned: {res['unowned_applications']})") + print(f"Resumes: {res['resumes_count']} (migrated: {res.get('migrated_resumes', 0)}, unowned: {res['unowned_resumes']})") + print(f"Attachments: {res['attachments_count']} (migrated: {res.get('migrated_attachments', 0)})") + print(f"User Settings: {res['user_settings_count']} (migrated: {res.get('migrated_user_settings', 0)})") + print(f"Global Settings: {res['global_settings_count']} (migrated: {res.get('migrated_global_settings', 0)})") + print(f"Provider Profiles: {res['provider_profiles_count']} (migrated: {res.get('migrated_provider_profiles', 0)})") + if res.get("errors"): + print("Errors encountered:") + for err in res["errors"]: + print(f" - {err}") except Exception as e: print(f"Migration error: {e}", file=sys.stderr) sys.exit(1) diff --git a/tests/test_migration.py b/tests/test_migration.py index 1a36ec6..c8b88b9 100644 --- a/tests/test_migration.py +++ b/tests/test_migration.py @@ -84,6 +84,20 @@ def create_sample_source_db(db_path: Path): value TEXT NOT NULL ) """) + cursor.execute(""" + CREATE TABLE provider_profiles ( + id TEXT PRIMARY KEY, + user_id TEXT NOT NULL, + name TEXT NOT NULL, + api_base_url TEXT, + model_name TEXT, + api_key_encrypted TEXT, + key_suffix TEXT, + is_active INTEGER DEFAULT 0, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL + ) + """) # Seed data: 2 users cursor.execute( @@ -120,7 +134,7 @@ def create_sample_source_db(db_path: Path): ("res-1", "user-1", "Primary Resume", "Experienced Developer", "resumes/user-1/uuid_resume.pdf", "att-1", "2026-01-01T00:00:00", "2026-01-01T00:00:00"), ) - # User Settings & Global Settings + # User Settings & Global Settings & Provider Profiles cursor.execute( "INSERT INTO user_settings VALUES (?, ?, ?, ?, ?)", ("user-1", "model_name", "gemini-2.0-flash", "2026-01-01T00:00:00", "2026-01-01T00:00:00"), @@ -129,6 +143,10 @@ def create_sample_source_db(db_path: Path): "INSERT INTO settings VALUES (?, ?)", ("system_version", "1.0"), ) + cursor.execute( + "INSERT INTO provider_profiles VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + ("prof-1", "user-1", "My OpenAI", "https://api.openai.com/v1", "gpt-4o", "enc_key", "...1234", 1, "2026-01-01T00:00:00", "2026-01-01T00:00:00"), + ) conn.commit() conn.close() @@ -147,6 +165,9 @@ def test_dry_run_reports_counts_and_leaves_destination_empty(tmp_path): assert result["unowned_applications"] == 1 assert result["resumes_count"] == 1 assert result["attachments_count"] == 1 + assert result["user_settings_count"] == 1 + assert result["global_settings_count"] == 1 + assert result["provider_profiles_count"] == 1 # Destination DB should not exist or be empty if dest_db.exists(): @@ -164,6 +185,10 @@ def test_apply_migrates_all_records_and_relationships(tmp_path): result = run_migration(source_path=str(source_db), dest_path=str(dest_db), dry_run=False) assert result["success"] is True assert result["dry_run"] is False + assert result["migrated_user_settings"] == 1 + assert result["migrated_global_settings"] == 1 + assert result["migrated_provider_profiles"] == 1 + assert result["errors"] == [] # Connect to dest and verify records conn = sqlite3.connect(str(dest_db)) @@ -187,6 +212,18 @@ def test_apply_migrates_all_records_and_relationships(tmp_path): assert len(resumes) == 1 assert resumes[0] == ("res-1", "user-1", "att-1") + user_settings = cursor.execute("SELECT user_id, key, value FROM user_settings").fetchall() + assert len(user_settings) == 1 + assert user_settings[0] == ("user-1", "model_name", "gemini-2.0-flash") + + global_settings = cursor.execute("SELECT key, value FROM settings").fetchall() + assert len(global_settings) == 1 + assert global_settings[0] == ("system_version", "1.0") + + profiles = cursor.execute("SELECT id, user_id, name, model_name FROM provider_profiles").fetchall() + assert len(profiles) == 1 + assert profiles[0] == ("prof-1", "user-1", "My OpenAI", "gpt-4o") + # Ledger exists ledger = cursor.execute("SELECT count(*) FROM _migration_ledger").fetchone()[0] assert ledger > 0 @@ -210,6 +247,9 @@ def test_apply_is_idempotent(tmp_path): assert cursor.execute("SELECT count(*) FROM users").fetchone()[0] == 2 assert cursor.execute("SELECT count(*) FROM applications").fetchone()[0] == 3 assert cursor.execute("SELECT count(*) FROM resumes").fetchone()[0] == 1 + assert cursor.execute("SELECT count(*) FROM user_settings").fetchone()[0] == 1 + assert cursor.execute("SELECT count(*) FROM settings").fetchone()[0] == 1 + assert cursor.execute("SELECT count(*) FROM provider_profiles").fetchone()[0] == 1 conn.close() @@ -218,3 +258,50 @@ def test_source_equals_destination_is_rejected(tmp_path): Path(same_db).touch() with pytest.raises(ValueError, match="Source and destination cannot be identical"): run_migration(source_path=same_db, dest_path=same_db, dry_run=False) + + +def test_postgres_destination_initialization(monkeypatch, tmp_path): + source_db = tmp_path / "source.db" + create_sample_source_db(source_db) + + captured = {} + + class MockStorageService: + def __init__(self, db_path=None, force_sqlite=False, database_url=None): + captured["db_path"] = db_path + captured["force_sqlite"] = force_sqlite + captured["database_url"] = database_url + self.is_postgres = True + + def _get_cursor(self): + from contextlib import contextmanager + + @contextmanager + def _cur(): + class DummyCursor: + def execute(self, *args, **kwargs): + pass + def fetchall(self): + return [] + def fetchone(self): + return None + yield DummyCursor() + return _cur() + + def _format_sql(self, sql): + return sql + + def close(self): + captured["closed"] = True + + monkeypatch.setattr("backend.migrate.StorageService", MockStorageService) + + result = run_migration( + source_path=str(source_db), + dest_path="postgresql://user:pass@host:5432/testdb", + dry_run=False, + ) + assert result["success"] is True + assert captured["database_url"] == "postgresql://user:pass@host:5432/testdb" + assert captured["db_path"] is None + assert captured["closed"] is True From fa46b5d70b74ccebe28be3dd33f645dec859bcfd Mon Sep 17 00:00:00 2001 From: Sourav Chandhok <114119117+souravC01@users.noreply.github.com> Date: Sat, 12 Sep 2026 00:06:38 -0400 Subject: [PATCH 09/14] feat(ui): Add duplicate job URL inline notification and responsive Kanban layout --- frontend/src/components/JobAnalyzer.jsx | 32 ++++++++++++++++------ frontend/src/components/KanbanBoard.jsx | 4 +-- frontend/src/test/JobAnalyzer.test.jsx | 36 +++++++++++++++++++++++++ 3 files changed, 62 insertions(+), 10 deletions(-) create mode 100644 frontend/src/test/JobAnalyzer.test.jsx diff --git a/frontend/src/components/JobAnalyzer.jsx b/frontend/src/components/JobAnalyzer.jsx index f96b03f..5dd2a6b 100644 --- a/frontend/src/components/JobAnalyzer.jsx +++ b/frontend/src/components/JobAnalyzer.jsx @@ -56,6 +56,12 @@ export default function JobAnalyzer({ }) ); + const matchingAppByUrl = Boolean(urlInput.trim()) && (applications || []).find((app) => { + const cleanInput = urlInput.trim().replace(/\/+$/, '').toLowerCase(); + const appUrl = (app.url || '').trim().replace(/\/+$/, '').toLowerCase(); + return cleanInput && appUrl && cleanInput === appUrl; + }); + const handleAnalyze = async (e) => { e?.preventDefault(); setError(''); @@ -123,14 +129,24 @@ export default function JobAnalyzer({
- setUrlInput(e.target.value)} - disabled={loading} - className="input-corporate flex-1 text-sm text-[#000000] placeholder:text-[#666666] h-11" - /> +
+ setUrlInput(e.target.value)} + disabled={loading} + className="input-corporate w-full text-sm text-[#000000] placeholder:text-[#666666] h-11" + /> + {matchingAppByUrl && ( +
+ Already tracked: + {matchingAppByUrl.company} + ({matchingAppByUrl.role}) + [{matchingAppByUrl.status}] +
+ )} +