diff --git a/templates/setup.html b/templates/setup.html
index 904ffb77..e5392aa8 100644
--- a/templates/setup.html
+++ b/templates/setup.html
@@ -36,6 +36,7 @@
.ptgl-row:hover { border-color:rgba(99,102,241,.25); }
.ptgl-row.disabled-row { opacity:.45; }
+ {% include "_csrf.html" %}
diff --git a/templates/website_hub.html b/templates/website_hub.html
index 38324996..605fc4ff 100644
--- a/templates/website_hub.html
+++ b/templates/website_hub.html
@@ -30,6 +30,7 @@
.toggle-switch span { position:relative; display:inline-block; width:2.75rem;
height:1.5rem; background:#d1d5db; border-radius:9999px; transition:.2s; cursor:pointer; }
+ {% include "_csrf.html" %}
diff --git a/tests/conftest.py b/tests/conftest.py
index 9d32f37d..36f91c7d 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -1,15 +1,18 @@
"""
conftest.py -- pytest fixtures for Post-Pilot.
-Forces SQLite in-memory so no real DB is needed to run tests.
-Disables CSRF so form posts work in tests.
+Forces SQLite so no real DB is needed to run tests.
+Disables CSRF so form posts work in tests (unless a test re-enables it).
"""
import os
+from unittest.mock import MagicMock, patch
-# Set env before any app imports
-os.environ.setdefault('DATABASE_URL', '')
-os.environ.setdefault('DATABASE_PATH', 'test_postpilot.db')
+# Set env before any app imports.
+# Force a file-backed SQLite DB — CI previously set DATABASE_PATH=:memory:, which
+# gives each connection a separate empty DB and breaks platform_tokens init.
+os.environ['DATABASE_URL'] = ''
+os.environ['DATABASE_PATH'] = 'test_postpilot.db'
try:
if os.path.exists('test_postpilot.db'):
os.remove('test_postpilot.db')
@@ -19,6 +22,7 @@
os.environ.setdefault('TOKEN_ENCRYPTION_KEY', 'nZeXTE9lxj2Vt5nUWGvfeVacq9+sz5XPLBuiA9wqjy4=')
os.environ.setdefault('FLASK_ENV', 'testing')
+os.environ.setdefault('APP_ENV', 'development')
os.environ.setdefault('OPENAI_API_KEY', 'dummy')
os.environ.setdefault('STRIPE_SECRET_KEY', 'dummy')
os.environ.setdefault('FACEBOOK_APP_ID', 'dummy')
@@ -40,39 +44,68 @@ def app():
'LOGIN_DISABLED': False,
})
with flask_app.app.app_context():
- flask_app.init_db()
+ from modules.auth_manager import init_db as auth_init_db
+ auth_init_db()
+ from modules.database import get_db
+ from modules.api_manager import CREATE_API_KEYS_TABLE
+ db = get_db()
+ db.execute(CREATE_API_KEYS_TABLE)
+ db.commit()
yield flask_app.app
try:
import gc
- gc.collect() # Force release of any open file handles
+ gc.collect()
if os.path.exists('test_postpilot.db'):
os.remove('test_postpilot.db')
except Exception:
pass
-
@pytest.fixture()
def client(app):
return app.test_client()
+def _make_mock_user(uid='00000000-0000-4000-8000-000000000001',
+ email='testuser@postpilot.dev',
+ tier='starter'):
+ user = MagicMock()
+ user.is_authenticated = True
+ user.is_active = True
+ user.is_anonymous = False
+ user.id = uid
+ user.email = email
+ user.subscription_tier = tier
+ user.plan = tier
+ user.get_id = lambda: uid
+ return user
+
+
@pytest.fixture()
-def registered_user(client):
- email = 'testuser@postpilot.dev'
- password = 'TestPass123!'
- client.post('/register', data={
- 'email': email,
- 'password': password,
- 'display_name': 'Test User',
- }, follow_redirects=True)
- return {'email': email, 'password': password}
+def registered_user():
+ return {
+ 'email': 'testuser@postpilot.dev',
+ 'password': 'unused',
+ 'id': '00000000-0000-4000-8000-000000000001',
+ }
@pytest.fixture()
-def logged_in_client(client, registered_user):
- client.post('/login', data={
- 'email': registered_user['email'],
- 'password': registered_user['password'],
- }, follow_redirects=True)
- return client
+def logged_in_client(client, registered_user, app):
+ """
+ Authenticated test client. Magic-link auth has no password login,
+ so we stub flask-login's user_loader for the session.
+ """
+ user = _make_mock_user(registered_user['id'], registered_user['email'], 'starter')
+
+ def _loader(uid):
+ if str(uid) == registered_user['id']:
+ return user
+ return None
+
+ with patch.object(app.login_manager, '_user_callback', _loader):
+ with client.session_transaction() as sess:
+ sess['_user_id'] = registered_user['id']
+ sess['_fresh'] = True
+ client._test_user = user
+ yield client
diff --git a/tests/test_p0_fixes.py b/tests/test_p0_fixes.py
index 74457c99..ef690d40 100644
--- a/tests/test_p0_fixes.py
+++ b/tests/test_p0_fixes.py
@@ -35,30 +35,32 @@ def test_free_user_blocked_on_publish_multi_platform(self, logged_in_client):
A free-tier user selecting more platforms than allowed should get 403
from /api/publish, not just /api/push_all.
"""
- with patch('modules.plan_guard.check_platform_limit', return_value=(False, 1)):
+ with patch('blueprints.api.check_platform_limit', return_value=(False, 1)):
resp = logged_in_client.post(
'/api/publish',
json={
'caption': 'Test post',
'content_type': 'text',
- 'platforms': ['fb', 'ig', 'tt'],
+ # fb/web need no media — avoid validator 400 before plan check
+ 'platforms': ['fb', 'web', 'gb'],
},
content_type='application/json',
)
assert resp.status_code == 403
data = resp.get_json()
- err_msg = data['error']['message'] if isinstance(data['error'], dict) else data['error']
- assert 'plan' in err_msg.lower() or 'limit' in err_msg.lower()
+ err = data.get('error', '')
+ err_msg = err['message'] if isinstance(err, dict) else str(err)
+ assert 'plan' in err_msg.lower() or 'limit' in err_msg.lower() or 'platform' in err_msg.lower()
def test_free_user_blocked_on_push_all_multi_platform(self, logged_in_client):
"""Same check on /api/push_all."""
- with patch('modules.plan_guard.check_platform_limit', return_value=(False, 1)):
+ with patch('blueprints.api.check_platform_limit', return_value=(False, 1)):
resp = logged_in_client.post(
'/api/push_all',
json={
'caption': 'Test post',
'content_type': 'text',
- 'platforms': ['fb', 'ig', 'tt'],
+ 'platforms': ['fb', 'web', 'gb'],
},
content_type='application/json',
)
@@ -66,23 +68,14 @@ def test_free_user_blocked_on_push_all_multi_platform(self, logged_in_client):
data = resp.get_json()
assert data['success'] is False
- def test_allowed_user_not_blocked(self, client, registered_user):
+ def test_allowed_user_not_blocked(self, logged_in_client):
"""When check_platform_limit returns allowed=True, the request proceeds past the guard."""
- # Log in the user
- client.post('/login', data={
- 'email': registered_user['email'],
- 'password': registered_user['password'],
- }, follow_redirects=True)
-
- mock_user = MagicMock()
- mock_user.subscription_tier = 'starter'
- mock_user.id = 'dummy-id'
-
- with patch('modules.plan_guard.current_user', mock_user), \
- patch('modules.plan_guard.check_platform_limit', return_value=(True, 3)), \
- patch('modules.publisher.UniversalPublisher.push_all', return_value={'fb': {'success': True}}), \
- patch('modules.user_manager.UserManager.log_post', return_value=None):
- resp = client.post(
+ with patch('blueprints.api.check_platform_limit', return_value=(True, 3)), \
+ patch('blueprints.api.UniversalPublisher') as Pub, \
+ patch('blueprints.api.UserManager.log_post', return_value=None), \
+ patch('blueprints.api._get_tokens', return_value={}):
+ Pub.return_value.push_all.return_value = {'fb': {'success': True}}
+ resp = logged_in_client.post(
'/api/publish',
json={
'caption': 'Hello world',
@@ -204,19 +197,34 @@ class TestPostHistoryLimit:
def test_non_integer_limit_returns_200(self, logged_in_client):
"""A non-integer ?limit= should fall back to default 20 and return 200."""
- resp = logged_in_client.get('/api/post_history?limit=abc')
+ from unittest.mock import MagicMock, patch
+ with patch('modules.database.get_db') as gdb:
+ mock_db = MagicMock()
+ mock_db.execute.return_value.fetchall.return_value = []
+ gdb.return_value = mock_db
+ resp = logged_in_client.get('/api/post_history?limit=abc')
assert resp.status_code == 200
data = resp.get_json()
assert data['success'] is True
def test_negative_limit_is_clamped(self, logged_in_client):
"""A negative limit should not crash; posts list is returned."""
- resp = logged_in_client.get('/api/post_history?limit=-5')
+ from unittest.mock import MagicMock, patch
+ with patch('modules.database.get_db') as gdb:
+ mock_db = MagicMock()
+ mock_db.execute.return_value.fetchall.return_value = []
+ gdb.return_value = mock_db
+ resp = logged_in_client.get('/api/post_history?limit=-5')
assert resp.status_code == 200
def test_over_max_limit_is_clamped_to_100(self, logged_in_client):
"""Limits above 100 should be silently clamped to 100."""
- resp = logged_in_client.get('/api/post_history?limit=9999')
+ from unittest.mock import MagicMock, patch
+ with patch('modules.database.get_db') as gdb:
+ mock_db = MagicMock()
+ mock_db.execute.return_value.fetchall.return_value = []
+ gdb.return_value = mock_db
+ resp = logged_in_client.get('/api/post_history?limit=9999')
assert resp.status_code == 200
diff --git a/tests/test_security_c1_c6.py b/tests/test_security_c1_c6.py
new file mode 100644
index 00000000..52628aec
--- /dev/null
+++ b/tests/test_security_c1_c6.py
@@ -0,0 +1,137 @@
+"""
+tests/test_security_c1_c6.py
+Regression tests for critical security fixes C1–C6.
+"""
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+
+# ---------------------------------------------------------------------------
+# C2 — /v1 IDOR: user API keys cannot act as another user_id
+# ---------------------------------------------------------------------------
+
+class TestV1UserIdBinding:
+
+ def test_resolve_scoped_user_id_binds_api_key_owner(self, app):
+ from modules.api_manager import _resolve_scoped_user_id
+ with app.test_request_context():
+ from flask import g
+ g.api_key_row = {'id': 1}
+ g.api_user_id = 'key-owner'
+ assert _resolve_scoped_user_id('attacker-supplied') == 'key-owner'
+
+ def test_resolve_scoped_user_id_allows_srn_explicit(self, app):
+ from modules.api_manager import _resolve_scoped_user_id
+ with app.test_request_context():
+ from flask import g
+ g.api_key_row = None
+ g.api_user_id = None
+ assert _resolve_scoped_user_id('srn-target') == 'srn-target'
+
+
+# ---------------------------------------------------------------------------
+# C1 — CSRF enforced on session API when enabled
+# ---------------------------------------------------------------------------
+
+class TestCSRFEnforced:
+
+ def test_publish_without_csrf_rejected_when_enabled(self, app, registered_user):
+ app.config['WTF_CSRF_ENABLED'] = True
+ try:
+ user = MagicMock()
+ user.is_authenticated = True
+ user.is_active = True
+ user.is_anonymous = False
+ user.id = registered_user['id']
+ user.subscription_tier = 'starter'
+ user.get_id = lambda: registered_user['id']
+
+ def _loader(uid):
+ return user if str(uid) == registered_user['id'] else None
+
+ client = app.test_client()
+ with patch.object(app.login_manager, '_user_callback', _loader):
+ with client.session_transaction() as sess:
+ sess['_user_id'] = registered_user['id']
+ sess['_fresh'] = True
+ resp = client.post(
+ '/api/publish',
+ json={
+ 'caption': 'csrf test',
+ 'content_type': 'text',
+ 'platforms': ['fb'],
+ },
+ content_type='application/json',
+ )
+ assert resp.status_code == 400
+ finally:
+ app.config['WTF_CSRF_ENABLED'] = False
+
+
+# ---------------------------------------------------------------------------
+# C6 — setup_tokens always gone (OAuth-only; matches main Wave A)
+# ---------------------------------------------------------------------------
+
+class TestSetupTokensLocked:
+
+ def test_setup_tokens_gone_in_production(self, logged_in_client, monkeypatch):
+ monkeypatch.setenv('APP_ENV', 'production')
+ monkeypatch.setenv('FLASK_ENV', 'production')
+ resp = logged_in_client.post(
+ '/api/setup_tokens',
+ json={'tokens': {'facebook_token': 'stolen'}},
+ content_type='application/json',
+ )
+ assert resp.status_code == 410
+ data = resp.get_json()
+ assert data['success'] is False
+ assert data['error']['code'] == 'GONE'
+
+ def test_setup_tokens_gone_in_dev(self, logged_in_client, monkeypatch):
+ monkeypatch.setenv('APP_ENV', 'development')
+ monkeypatch.setenv('FLASK_ENV', 'development')
+ resp = logged_in_client.post(
+ '/api/setup_tokens',
+ json={'tokens': {'tiktok_token': 'dev-token'}},
+ content_type='application/json',
+ )
+ assert resp.status_code == 410
+ assert resp.get_json()['success'] is False
+
+
+# ---------------------------------------------------------------------------
+# C3 — TOKEN_ENCRYPTION_KEY fail-closed helper
+# ---------------------------------------------------------------------------
+
+class TestEncryptionKeyPolicy:
+
+ def test_is_production_helper(self, monkeypatch):
+ from modules.auth_manager import _is_production
+ monkeypatch.delenv('VERCEL_ENV', raising=False)
+ monkeypatch.setenv('APP_ENV', 'development')
+ monkeypatch.setenv('FLASK_ENV', 'development')
+ assert _is_production() is False
+ monkeypatch.setenv('APP_ENV', 'production')
+ assert _is_production() is True
+ monkeypatch.setenv('APP_ENV', 'development')
+ monkeypatch.setenv('VERCEL_ENV', 'preview')
+ assert _is_production() is False
+ monkeypatch.setenv('VERCEL_ENV', 'production')
+ assert _is_production() is True
+
+
+# ---------------------------------------------------------------------------
+# C5 — get_db import path
+# ---------------------------------------------------------------------------
+
+class TestGetDbImport:
+
+ def test_modules_database_get_db_callable(self, app):
+ from modules.database import get_db
+ with app.app_context():
+ db = get_db()
+ assert db is not None
+ row = db.execute('SELECT 1 AS n').fetchone()
+ assert row is not None
diff --git a/tests/test_smoke.py b/tests/test_smoke.py
index 45d6b4e9..21963d25 100644
--- a/tests/test_smoke.py
+++ b/tests/test_smoke.py
@@ -31,17 +31,16 @@ class TestAuthFlow:
def test_register_new_user(self, client):
resp = client.post('/register', data={
'email': 'brand-new@example.com',
- 'password': 'SecurePass1!',
'display_name': 'Brand New',
}, follow_redirects=True)
assert resp.status_code == 200
- def test_login_wrong_password(self, client):
+ def test_login_unknown_email_does_not_500(self, client):
+ """Magic-link login: unknown email should not crash (link send may fail in CI)."""
resp = client.post('/login', data={
- 'email': 'nobody@example.com',
- 'password': 'wrongpassword',
+ 'email': 'nobody@example.com',
}, follow_redirects=True)
- assert b'Invalid email or password' in resp.data
+ assert resp.status_code == 200
def test_protected_redirects_anonymous(self, client):
resp = client.get('/dashboard')
@@ -57,7 +56,12 @@ def test_billing(self, logged_in_client):
assert logged_in_client.get('/billing').status_code == 200
def test_post_history_returns_list(self, logged_in_client):
- resp = logged_in_client.get('/api/post_history')
+ from unittest.mock import MagicMock, patch
+ with patch('modules.database.get_db') as gdb:
+ mock_db = MagicMock()
+ mock_db.execute.return_value.fetchall.return_value = []
+ gdb.return_value = mock_db
+ resp = logged_in_client.get('/api/post_history')
assert resp.status_code == 200
data = resp.get_json()
assert data['success'] is True
diff --git a/vercel.json b/vercel.json
index ccbac275..93b9c7fc 100644
--- a/vercel.json
+++ b/vercel.json
@@ -9,11 +9,11 @@
"crons": [
{
"path": "/api/cron/generate",
- "schedule": "0 * * * *"
+ "schedule": "0 14 * * *"
},
{
"path": "/api/cron/publish",
- "schedule": "* * * * *"
+ "schedule": "0 15 * * *"
}
]
}