forked from claw/flaskpaste
pki: add minimal certificate authority
- CA generation with encrypted private key storage (AES-256-GCM) - Client certificate issuance with configurable validity - Certificate revocation with status tracking - SHA1 fingerprint integration with existing mTLS auth - API endpoints: /pki/status, /pki/ca, /pki/issue, /pki/revoke - CLI commands: fpaste pki status/issue/revoke - Comprehensive test coverage
This commit is contained in:
Username
500
tests/test_paste_options.py
Normal file
500
tests/test_paste_options.py
Normal file
@@ -0,0 +1,500 @@
|
||||
"""Tests for burn-after-read and custom expiry features."""
|
||||
|
||||
import time
|
||||
|
||||
import pytest
|
||||
|
||||
from app import create_app
|
||||
from app.database import cleanup_expired_pastes
|
||||
|
||||
|
||||
class TestBurnAfterRead:
|
||||
"""Test burn-after-read paste functionality."""
|
||||
|
||||
@pytest.fixture
|
||||
def app(self):
|
||||
"""Create app for burn-after-read tests."""
|
||||
return create_app("testing")
|
||||
|
||||
@pytest.fixture
|
||||
def client(self, app):
|
||||
"""Create test client."""
|
||||
return app.test_client()
|
||||
|
||||
def test_create_burn_paste(self, client):
|
||||
"""Creating a burn-after-read paste should succeed."""
|
||||
response = client.post(
|
||||
"/",
|
||||
data=b"secret message",
|
||||
headers={"X-Burn-After-Read": "true"},
|
||||
)
|
||||
assert response.status_code == 201
|
||||
data = response.get_json()
|
||||
assert data["burn_after_read"] is True
|
||||
|
||||
def test_burn_paste_deleted_after_raw_get(self, client):
|
||||
"""Burn paste should be deleted after first GET /raw."""
|
||||
# Create burn paste
|
||||
response = client.post(
|
||||
"/",
|
||||
data=b"one-time secret",
|
||||
headers={"X-Burn-After-Read": "true"},
|
||||
)
|
||||
paste_id = response.get_json()["id"]
|
||||
|
||||
# First GET should succeed
|
||||
response = client.get(f"/{paste_id}/raw")
|
||||
assert response.status_code == 200
|
||||
assert response.data == b"one-time secret"
|
||||
assert response.headers.get("X-Burn-After-Read") == "true"
|
||||
|
||||
# Second GET should fail (paste deleted)
|
||||
response = client.get(f"/{paste_id}/raw")
|
||||
assert response.status_code == 404
|
||||
|
||||
def test_burn_paste_metadata_does_not_trigger_burn(self, client):
|
||||
"""GET metadata should not delete burn paste."""
|
||||
# Create burn paste
|
||||
response = client.post(
|
||||
"/",
|
||||
data=b"secret",
|
||||
headers={"X-Burn-After-Read": "true"},
|
||||
)
|
||||
paste_id = response.get_json()["id"]
|
||||
|
||||
# Metadata GET should succeed and show burn flag
|
||||
response = client.get(f"/{paste_id}")
|
||||
assert response.status_code == 200
|
||||
data = response.get_json()
|
||||
assert data["burn_after_read"] is True
|
||||
|
||||
# Paste should still exist
|
||||
response = client.get(f"/{paste_id}")
|
||||
assert response.status_code == 200
|
||||
|
||||
# Raw GET should delete it
|
||||
response = client.get(f"/{paste_id}/raw")
|
||||
assert response.status_code == 200
|
||||
|
||||
# Now it's gone
|
||||
response = client.get(f"/{paste_id}")
|
||||
assert response.status_code == 404
|
||||
|
||||
def test_head_does_not_trigger_burn(self, client):
|
||||
"""HEAD request should not delete burn paste."""
|
||||
# Create burn paste
|
||||
response = client.post(
|
||||
"/",
|
||||
data=b"secret",
|
||||
headers={"X-Burn-After-Read": "true"},
|
||||
)
|
||||
paste_id = response.get_json()["id"]
|
||||
|
||||
# HEAD should succeed
|
||||
response = client.head(f"/{paste_id}/raw")
|
||||
assert response.status_code == 200
|
||||
|
||||
# Paste should still exist
|
||||
response = client.get(f"/{paste_id}/raw")
|
||||
assert response.status_code == 200
|
||||
|
||||
def test_burn_header_variations(self, client):
|
||||
"""Different true values for X-Burn-After-Read should work."""
|
||||
for value in ["true", "TRUE", "1", "yes", "YES"]:
|
||||
response = client.post(
|
||||
"/",
|
||||
data=b"content",
|
||||
headers={"X-Burn-After-Read": value},
|
||||
)
|
||||
data = response.get_json()
|
||||
assert data.get("burn_after_read") is True, f"Failed for value: {value}"
|
||||
|
||||
def test_burn_header_false_values(self, client):
|
||||
"""False values should not enable burn-after-read."""
|
||||
for value in ["false", "0", "no", ""]:
|
||||
response = client.post(
|
||||
"/",
|
||||
data=b"content",
|
||||
headers={"X-Burn-After-Read": value},
|
||||
)
|
||||
data = response.get_json()
|
||||
assert "burn_after_read" not in data, f"Should not be burn for: {value}"
|
||||
|
||||
|
||||
class TestCustomExpiry:
|
||||
"""Test custom expiry functionality."""
|
||||
|
||||
@pytest.fixture
|
||||
def app(self):
|
||||
"""Create app with short max expiry for testing."""
|
||||
app = create_app("testing")
|
||||
app.config["MAX_EXPIRY_SECONDS"] = 3600 # 1 hour max
|
||||
app.config["PASTE_EXPIRY_SECONDS"] = 60 # 1 minute default
|
||||
return app
|
||||
|
||||
@pytest.fixture
|
||||
def client(self, app):
|
||||
"""Create test client."""
|
||||
return app.test_client()
|
||||
|
||||
def test_create_paste_with_custom_expiry(self, client):
|
||||
"""Creating a paste with X-Expiry should set expires_at."""
|
||||
response = client.post(
|
||||
"/",
|
||||
data=b"temporary content",
|
||||
headers={"X-Expiry": "300"}, # 5 minutes
|
||||
)
|
||||
assert response.status_code == 201
|
||||
data = response.get_json()
|
||||
assert "expires_at" in data
|
||||
# Should be approximately now + 300
|
||||
now = int(time.time())
|
||||
assert abs(data["expires_at"] - (now + 300)) < 5
|
||||
|
||||
def test_custom_expiry_capped_at_max(self, client):
|
||||
"""Custom expiry should be capped at MAX_EXPIRY_SECONDS."""
|
||||
response = client.post(
|
||||
"/",
|
||||
data=b"content",
|
||||
headers={"X-Expiry": "999999"}, # Way more than max
|
||||
)
|
||||
assert response.status_code == 201
|
||||
data = response.get_json()
|
||||
assert "expires_at" in data
|
||||
# Should be capped at 3600 seconds from now
|
||||
now = int(time.time())
|
||||
assert abs(data["expires_at"] - (now + 3600)) < 5
|
||||
|
||||
def test_expiry_shown_in_metadata(self, client):
|
||||
"""Custom expiry should appear in paste metadata."""
|
||||
response = client.post(
|
||||
"/",
|
||||
data=b"content",
|
||||
headers={"X-Expiry": "600"},
|
||||
)
|
||||
paste_id = response.get_json()["id"]
|
||||
|
||||
response = client.get(f"/{paste_id}")
|
||||
data = response.get_json()
|
||||
assert "expires_at" in data
|
||||
|
||||
def test_invalid_expiry_ignored(self, client):
|
||||
"""Invalid X-Expiry values should be ignored."""
|
||||
for value in ["invalid", "-100", "0", ""]:
|
||||
response = client.post(
|
||||
"/",
|
||||
data=b"content",
|
||||
headers={"X-Expiry": value},
|
||||
)
|
||||
assert response.status_code == 201
|
||||
data = response.get_json()
|
||||
assert "expires_at" not in data, f"Should not have expiry for: {value}"
|
||||
|
||||
def test_paste_without_custom_expiry(self, client):
|
||||
"""Paste without X-Expiry should not have expires_at."""
|
||||
response = client.post("/", data=b"content")
|
||||
assert response.status_code == 201
|
||||
data = response.get_json()
|
||||
assert "expires_at" not in data
|
||||
|
||||
|
||||
class TestExpiryCleanup:
|
||||
"""Test cleanup of expired pastes."""
|
||||
|
||||
@pytest.fixture
|
||||
def app(self):
|
||||
"""Create app with very short expiry for testing."""
|
||||
app = create_app("testing")
|
||||
app.config["PASTE_EXPIRY_SECONDS"] = 1 # 1 second default
|
||||
app.config["MAX_EXPIRY_SECONDS"] = 10
|
||||
return app
|
||||
|
||||
@pytest.fixture
|
||||
def client(self, app):
|
||||
"""Create test client."""
|
||||
return app.test_client()
|
||||
|
||||
def test_cleanup_custom_expired_paste(self, app, client):
|
||||
"""Paste with expired custom expiry should be cleaned up."""
|
||||
# Create paste with 1 second expiry
|
||||
response = client.post(
|
||||
"/",
|
||||
data=b"expiring soon",
|
||||
headers={"X-Expiry": "1"},
|
||||
)
|
||||
paste_id = response.get_json()["id"]
|
||||
|
||||
# Should exist immediately
|
||||
response = client.get(f"/{paste_id}")
|
||||
assert response.status_code == 200
|
||||
|
||||
# Wait for expiry
|
||||
time.sleep(2)
|
||||
|
||||
# Run cleanup
|
||||
with app.app_context():
|
||||
deleted = cleanup_expired_pastes()
|
||||
assert deleted >= 1
|
||||
|
||||
# Should be gone
|
||||
response = client.get(f"/{paste_id}")
|
||||
assert response.status_code == 404
|
||||
|
||||
def test_cleanup_respects_default_expiry(self, app, client):
|
||||
"""Paste without custom expiry should use default expiry."""
|
||||
# Create paste without custom expiry
|
||||
response = client.post("/", data=b"default expiry")
|
||||
paste_id = response.get_json()["id"]
|
||||
|
||||
# Wait for default expiry (1 second in test config)
|
||||
time.sleep(2)
|
||||
|
||||
# Run cleanup
|
||||
with app.app_context():
|
||||
deleted = cleanup_expired_pastes()
|
||||
assert deleted >= 1
|
||||
|
||||
# Should be gone
|
||||
response = client.get(f"/{paste_id}")
|
||||
assert response.status_code == 404
|
||||
|
||||
def test_cleanup_keeps_unexpired_paste(self, app, client):
|
||||
"""Paste with future custom expiry should not be cleaned up."""
|
||||
# Create paste with long expiry
|
||||
response = client.post(
|
||||
"/",
|
||||
data=b"not expiring soon",
|
||||
headers={"X-Expiry": "10"}, # 10 seconds
|
||||
)
|
||||
paste_id = response.get_json()["id"]
|
||||
|
||||
# Run cleanup immediately
|
||||
with app.app_context():
|
||||
cleanup_expired_pastes()
|
||||
|
||||
# Should still exist
|
||||
response = client.get(f"/{paste_id}")
|
||||
assert response.status_code == 200
|
||||
|
||||
|
||||
class TestCombinedOptions:
|
||||
"""Test combinations of burn-after-read and custom expiry."""
|
||||
|
||||
@pytest.fixture
|
||||
def app(self):
|
||||
"""Create app for combined tests."""
|
||||
return create_app("testing")
|
||||
|
||||
@pytest.fixture
|
||||
def client(self, app):
|
||||
"""Create test client."""
|
||||
return app.test_client()
|
||||
|
||||
def test_burn_and_expiry_together(self, client):
|
||||
"""Paste can have both burn-after-read and custom expiry."""
|
||||
response = client.post(
|
||||
"/",
|
||||
data=b"secret with expiry",
|
||||
headers={
|
||||
"X-Burn-After-Read": "true",
|
||||
"X-Expiry": "3600",
|
||||
},
|
||||
)
|
||||
assert response.status_code == 201
|
||||
data = response.get_json()
|
||||
assert data["burn_after_read"] is True
|
||||
assert "expires_at" in data
|
||||
|
||||
|
||||
class TestPasswordProtection:
|
||||
"""Test password-protected paste functionality."""
|
||||
|
||||
@pytest.fixture
|
||||
def app(self):
|
||||
"""Create app for password tests."""
|
||||
return create_app("testing")
|
||||
|
||||
@pytest.fixture
|
||||
def client(self, app):
|
||||
"""Create test client."""
|
||||
return app.test_client()
|
||||
|
||||
def test_create_password_protected_paste(self, client):
|
||||
"""Creating a password-protected paste should succeed."""
|
||||
response = client.post(
|
||||
"/",
|
||||
data=b"secret content",
|
||||
headers={"X-Paste-Password": "mypassword123"},
|
||||
)
|
||||
assert response.status_code == 201
|
||||
data = response.get_json()
|
||||
assert data["password_protected"] is True
|
||||
|
||||
def test_get_protected_paste_without_password(self, client):
|
||||
"""Accessing protected paste without password should return 401."""
|
||||
# Create protected paste
|
||||
response = client.post(
|
||||
"/",
|
||||
data=b"protected content",
|
||||
headers={"X-Paste-Password": "secret"},
|
||||
)
|
||||
paste_id = response.get_json()["id"]
|
||||
|
||||
# Try to access without password
|
||||
response = client.get(f"/{paste_id}")
|
||||
assert response.status_code == 401
|
||||
data = response.get_json()
|
||||
assert data["password_protected"] is True
|
||||
assert "Password required" in data["error"]
|
||||
|
||||
def test_get_protected_paste_with_wrong_password(self, client):
|
||||
"""Accessing protected paste with wrong password should return 403."""
|
||||
# Create protected paste
|
||||
response = client.post(
|
||||
"/",
|
||||
data=b"protected content",
|
||||
headers={"X-Paste-Password": "correctpassword"},
|
||||
)
|
||||
paste_id = response.get_json()["id"]
|
||||
|
||||
# Try with wrong password
|
||||
response = client.get(
|
||||
f"/{paste_id}",
|
||||
headers={"X-Paste-Password": "wrongpassword"},
|
||||
)
|
||||
assert response.status_code == 403
|
||||
data = response.get_json()
|
||||
assert "Invalid password" in data["error"]
|
||||
|
||||
def test_get_protected_paste_with_correct_password(self, client):
|
||||
"""Accessing protected paste with correct password should succeed."""
|
||||
password = "supersecret123"
|
||||
# Create protected paste
|
||||
response = client.post(
|
||||
"/",
|
||||
data=b"protected content",
|
||||
headers={"X-Paste-Password": password},
|
||||
)
|
||||
paste_id = response.get_json()["id"]
|
||||
|
||||
# Access with correct password
|
||||
response = client.get(
|
||||
f"/{paste_id}",
|
||||
headers={"X-Paste-Password": password},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = response.get_json()
|
||||
assert data["password_protected"] is True
|
||||
|
||||
def test_get_raw_protected_paste_without_password(self, client):
|
||||
"""Getting raw content without password should return 401."""
|
||||
response = client.post(
|
||||
"/",
|
||||
data=b"secret raw content",
|
||||
headers={"X-Paste-Password": "secret"},
|
||||
)
|
||||
paste_id = response.get_json()["id"]
|
||||
|
||||
response = client.get(f"/{paste_id}/raw")
|
||||
assert response.status_code == 401
|
||||
|
||||
def test_get_raw_protected_paste_with_correct_password(self, client):
|
||||
"""Getting raw content with correct password should succeed."""
|
||||
password = "mypassword"
|
||||
response = client.post(
|
||||
"/",
|
||||
data=b"secret raw content",
|
||||
headers={"X-Paste-Password": password},
|
||||
)
|
||||
paste_id = response.get_json()["id"]
|
||||
|
||||
response = client.get(
|
||||
f"/{paste_id}/raw",
|
||||
headers={"X-Paste-Password": password},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
assert response.data == b"secret raw content"
|
||||
|
||||
def test_password_too_long_rejected(self, client):
|
||||
"""Password longer than 1024 chars should be rejected."""
|
||||
long_password = "x" * 1025
|
||||
response = client.post(
|
||||
"/",
|
||||
data=b"content",
|
||||
headers={"X-Paste-Password": long_password},
|
||||
)
|
||||
assert response.status_code == 400
|
||||
data = response.get_json()
|
||||
assert "too long" in data["error"]
|
||||
|
||||
def test_unprotected_paste_accessible(self, client):
|
||||
"""Unprotected paste should be accessible without password."""
|
||||
response = client.post("/", data=b"public content")
|
||||
paste_id = response.get_json()["id"]
|
||||
|
||||
response = client.get(f"/{paste_id}")
|
||||
assert response.status_code == 200
|
||||
assert "password_protected" not in response.get_json()
|
||||
|
||||
def test_password_with_special_chars(self, client):
|
||||
"""Password with special characters should work."""
|
||||
password = "p@ssw0rd!#$%^&*()_+-=[]{}|;':\",./<>?"
|
||||
response = client.post(
|
||||
"/",
|
||||
data=b"special content",
|
||||
headers={"X-Paste-Password": password},
|
||||
)
|
||||
paste_id = response.get_json()["id"]
|
||||
|
||||
response = client.get(
|
||||
f"/{paste_id}",
|
||||
headers={"X-Paste-Password": password},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
|
||||
def test_password_with_unicode(self, client):
|
||||
"""Password with unicode characters should work."""
|
||||
password = "пароль密码🔐"
|
||||
response = client.post(
|
||||
"/",
|
||||
data=b"unicode content",
|
||||
headers={"X-Paste-Password": password},
|
||||
)
|
||||
paste_id = response.get_json()["id"]
|
||||
|
||||
response = client.get(
|
||||
f"/{paste_id}",
|
||||
headers={"X-Paste-Password": password},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
|
||||
def test_password_combined_with_burn(self, client):
|
||||
"""Password protection can be combined with burn-after-read."""
|
||||
password = "secret"
|
||||
response = client.post(
|
||||
"/",
|
||||
data=b"protected burn content",
|
||||
headers={
|
||||
"X-Paste-Password": password,
|
||||
"X-Burn-After-Read": "true",
|
||||
},
|
||||
)
|
||||
assert response.status_code == 201
|
||||
data = response.get_json()
|
||||
assert data["password_protected"] is True
|
||||
assert data["burn_after_read"] is True
|
||||
paste_id = data["id"]
|
||||
|
||||
# First access with password should succeed
|
||||
response = client.get(
|
||||
f"/{paste_id}/raw",
|
||||
headers={"X-Paste-Password": password},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
|
||||
# Second access should fail (burned)
|
||||
response = client.get(
|
||||
f"/{paste_id}/raw",
|
||||
headers={"X-Paste-Password": password},
|
||||
)
|
||||
assert response.status_code == 404
|
||||
371
tests/test_pki.py
Normal file
371
tests/test_pki.py
Normal file
@@ -0,0 +1,371 @@
|
||||
"""Tests for PKI (Certificate Authority) functionality."""
|
||||
|
||||
from datetime import UTC
|
||||
|
||||
import pytest
|
||||
|
||||
from app.pki import reset_pki
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def reset_pki_state(app):
|
||||
"""Reset PKI state and clear PKI database tables before each test."""
|
||||
reset_pki()
|
||||
|
||||
# Clear PKI tables in database
|
||||
with app.app_context():
|
||||
from app.database import get_db
|
||||
|
||||
db = get_db()
|
||||
db.execute("DELETE FROM issued_certificates")
|
||||
db.execute("DELETE FROM certificate_authority")
|
||||
db.commit()
|
||||
|
||||
yield
|
||||
reset_pki()
|
||||
|
||||
|
||||
class TestPKIStatus:
|
||||
"""Test GET /pki endpoint."""
|
||||
|
||||
def test_pki_status_when_enabled(self, client):
|
||||
"""PKI status shows enabled with no CA initially."""
|
||||
response = client.get("/pki")
|
||||
assert response.status_code == 200
|
||||
data = response.get_json()
|
||||
assert data["enabled"] is True
|
||||
assert data["ca_exists"] is False
|
||||
assert "hint" in data
|
||||
|
||||
def test_pki_status_after_ca_generation(self, client):
|
||||
"""PKI status shows CA info after generation."""
|
||||
# Generate CA first
|
||||
client.post("/pki/ca", json={"common_name": "Test CA"})
|
||||
|
||||
response = client.get("/pki")
|
||||
assert response.status_code == 200
|
||||
data = response.get_json()
|
||||
assert data["enabled"] is True
|
||||
assert data["ca_exists"] is True
|
||||
assert data["common_name"] == "Test CA"
|
||||
assert "fingerprint_sha1" in data
|
||||
assert len(data["fingerprint_sha1"]) == 40
|
||||
|
||||
|
||||
class TestCAGeneration:
|
||||
"""Test POST /pki/ca endpoint."""
|
||||
|
||||
def test_generate_ca_success(self, client):
|
||||
"""CA can be generated with default name."""
|
||||
response = client.post("/pki/ca")
|
||||
assert response.status_code == 201
|
||||
data = response.get_json()
|
||||
assert data["message"] == "CA generated"
|
||||
assert data["common_name"] == "FlaskPaste CA"
|
||||
assert "fingerprint_sha1" in data
|
||||
assert "created_at" in data
|
||||
assert "expires_at" in data
|
||||
assert data["download"] == "/pki/ca.crt"
|
||||
|
||||
def test_generate_ca_custom_name(self, client):
|
||||
"""CA can be generated with custom name."""
|
||||
response = client.post("/pki/ca", json={"common_name": "My Custom CA"})
|
||||
assert response.status_code == 201
|
||||
data = response.get_json()
|
||||
assert data["common_name"] == "My Custom CA"
|
||||
|
||||
def test_generate_ca_twice_fails(self, client):
|
||||
"""CA cannot be generated twice."""
|
||||
# First generation succeeds
|
||||
response = client.post("/pki/ca")
|
||||
assert response.status_code == 201
|
||||
|
||||
# Second generation fails
|
||||
response = client.post("/pki/ca")
|
||||
assert response.status_code == 409
|
||||
data = response.get_json()
|
||||
assert "already exists" in data["error"]
|
||||
|
||||
|
||||
class TestCADownload:
|
||||
"""Test GET /pki/ca.crt endpoint."""
|
||||
|
||||
def test_download_ca_not_initialized(self, client):
|
||||
"""Download fails when no CA exists."""
|
||||
response = client.get("/pki/ca.crt")
|
||||
assert response.status_code == 404
|
||||
|
||||
def test_download_ca_success(self, client):
|
||||
"""CA certificate can be downloaded."""
|
||||
# Generate CA first
|
||||
client.post("/pki/ca", json={"common_name": "Test CA"})
|
||||
|
||||
response = client.get("/pki/ca.crt")
|
||||
assert response.status_code == 200
|
||||
assert response.content_type == "application/x-pem-file"
|
||||
assert b"-----BEGIN CERTIFICATE-----" in response.data
|
||||
assert b"-----END CERTIFICATE-----" in response.data
|
||||
|
||||
|
||||
class TestCertificateIssuance:
|
||||
"""Test POST /pki/issue endpoint."""
|
||||
|
||||
def test_issue_without_ca_fails(self, client):
|
||||
"""Issuance fails when no CA exists."""
|
||||
response = client.post("/pki/issue", json={"common_name": "alice"})
|
||||
assert response.status_code == 404
|
||||
|
||||
def test_issue_without_name_fails(self, client):
|
||||
"""Issuance fails without common_name."""
|
||||
client.post("/pki/ca")
|
||||
|
||||
response = client.post("/pki/issue", json={})
|
||||
assert response.status_code == 400
|
||||
assert "common_name required" in response.get_json()["error"]
|
||||
|
||||
def test_issue_certificate_success(self, client):
|
||||
"""Certificate issuance succeeds."""
|
||||
client.post("/pki/ca")
|
||||
|
||||
response = client.post("/pki/issue", json={"common_name": "alice"})
|
||||
assert response.status_code == 201
|
||||
data = response.get_json()
|
||||
assert data["message"] == "Certificate issued"
|
||||
assert data["common_name"] == "alice"
|
||||
assert "serial" in data
|
||||
assert "fingerprint_sha1" in data
|
||||
assert len(data["fingerprint_sha1"]) == 40
|
||||
assert "certificate_pem" in data
|
||||
assert "private_key_pem" in data
|
||||
assert "-----BEGIN CERTIFICATE-----" in data["certificate_pem"]
|
||||
assert "-----BEGIN PRIVATE KEY-----" in data["private_key_pem"]
|
||||
|
||||
def test_issue_multiple_certificates(self, client):
|
||||
"""Multiple certificates can be issued."""
|
||||
client.post("/pki/ca")
|
||||
|
||||
response1 = client.post("/pki/issue", json={"common_name": "alice"})
|
||||
response2 = client.post("/pki/issue", json={"common_name": "bob"})
|
||||
|
||||
assert response1.status_code == 201
|
||||
assert response2.status_code == 201
|
||||
|
||||
data1 = response1.get_json()
|
||||
data2 = response2.get_json()
|
||||
|
||||
# Different serials and fingerprints
|
||||
assert data1["serial"] != data2["serial"]
|
||||
assert data1["fingerprint_sha1"] != data2["fingerprint_sha1"]
|
||||
|
||||
|
||||
class TestCertificateListing:
|
||||
"""Test GET /pki/certs endpoint."""
|
||||
|
||||
def test_list_anonymous_empty(self, client):
|
||||
"""Anonymous users see empty list."""
|
||||
client.post("/pki/ca")
|
||||
|
||||
response = client.get("/pki/certs")
|
||||
assert response.status_code == 200
|
||||
data = response.get_json()
|
||||
assert data["certificates"] == []
|
||||
assert data["count"] == 0
|
||||
|
||||
def test_list_authenticated_sees_own(self, client):
|
||||
"""Authenticated users see certificates they issued."""
|
||||
client.post("/pki/ca")
|
||||
|
||||
# Issue certificate as authenticated user
|
||||
issuer_fingerprint = "a" * 40
|
||||
client.post(
|
||||
"/pki/issue",
|
||||
json={"common_name": "alice"},
|
||||
headers={"X-SSL-Client-SHA1": issuer_fingerprint},
|
||||
)
|
||||
|
||||
# List as same user
|
||||
response = client.get("/pki/certs", headers={"X-SSL-Client-SHA1": issuer_fingerprint})
|
||||
assert response.status_code == 200
|
||||
data = response.get_json()
|
||||
assert data["count"] == 1
|
||||
assert data["certificates"][0]["common_name"] == "alice"
|
||||
|
||||
|
||||
class TestCertificateRevocation:
|
||||
"""Test POST /pki/revoke/<serial> endpoint."""
|
||||
|
||||
def test_revoke_unauthenticated_fails(self, client):
|
||||
"""Revocation requires authentication."""
|
||||
client.post("/pki/ca")
|
||||
issue_resp = client.post("/pki/issue", json={"common_name": "alice"})
|
||||
serial = issue_resp.get_json()["serial"]
|
||||
|
||||
response = client.post(f"/pki/revoke/{serial}")
|
||||
assert response.status_code == 401
|
||||
|
||||
def test_revoke_unauthorized_fails(self, client):
|
||||
"""Revocation requires ownership."""
|
||||
client.post("/pki/ca")
|
||||
|
||||
# Issue as one user
|
||||
issue_resp = client.post(
|
||||
"/pki/issue", json={"common_name": "alice"}, headers={"X-SSL-Client-SHA1": "a" * 40}
|
||||
)
|
||||
serial = issue_resp.get_json()["serial"]
|
||||
|
||||
# Try to revoke as different user
|
||||
response = client.post(f"/pki/revoke/{serial}", headers={"X-SSL-Client-SHA1": "b" * 40})
|
||||
assert response.status_code == 403
|
||||
|
||||
def test_revoke_as_issuer_succeeds(self, client):
|
||||
"""Issuer can revoke certificate."""
|
||||
client.post("/pki/ca")
|
||||
|
||||
issuer = "a" * 40
|
||||
issue_resp = client.post(
|
||||
"/pki/issue", json={"common_name": "alice"}, headers={"X-SSL-Client-SHA1": issuer}
|
||||
)
|
||||
serial = issue_resp.get_json()["serial"]
|
||||
|
||||
response = client.post(f"/pki/revoke/{serial}", headers={"X-SSL-Client-SHA1": issuer})
|
||||
assert response.status_code == 200
|
||||
assert response.get_json()["message"] == "Certificate revoked"
|
||||
|
||||
def test_revoke_nonexistent_fails(self, client):
|
||||
"""Revoking nonexistent certificate fails."""
|
||||
client.post("/pki/ca")
|
||||
|
||||
response = client.post("/pki/revoke/0" * 32, headers={"X-SSL-Client-SHA1": "a" * 40})
|
||||
assert response.status_code == 404
|
||||
|
||||
def test_revoke_twice_fails(self, client):
|
||||
"""Certificate cannot be revoked twice."""
|
||||
client.post("/pki/ca")
|
||||
|
||||
issuer = "a" * 40
|
||||
issue_resp = client.post(
|
||||
"/pki/issue", json={"common_name": "alice"}, headers={"X-SSL-Client-SHA1": issuer}
|
||||
)
|
||||
serial = issue_resp.get_json()["serial"]
|
||||
|
||||
# First revocation succeeds
|
||||
response = client.post(f"/pki/revoke/{serial}", headers={"X-SSL-Client-SHA1": issuer})
|
||||
assert response.status_code == 200
|
||||
|
||||
# Second revocation fails
|
||||
response = client.post(f"/pki/revoke/{serial}", headers={"X-SSL-Client-SHA1": issuer})
|
||||
assert response.status_code == 409
|
||||
|
||||
|
||||
class TestRevocationIntegration:
|
||||
"""Test revocation affects authentication."""
|
||||
|
||||
def test_revoked_cert_treated_as_anonymous(self, client):
|
||||
"""Revoked certificate is treated as anonymous."""
|
||||
client.post("/pki/ca")
|
||||
|
||||
# Issue certificate
|
||||
issuer = "a" * 40
|
||||
issue_resp = client.post(
|
||||
"/pki/issue", json={"common_name": "alice"}, headers={"X-SSL-Client-SHA1": issuer}
|
||||
)
|
||||
cert_fingerprint = issue_resp.get_json()["fingerprint_sha1"]
|
||||
serial = issue_resp.get_json()["serial"]
|
||||
|
||||
# Create paste as authenticated user
|
||||
create_resp = client.post(
|
||||
"/", data=b"test content", headers={"X-SSL-Client-SHA1": cert_fingerprint}
|
||||
)
|
||||
assert create_resp.status_code == 201
|
||||
paste_id = create_resp.get_json()["id"]
|
||||
assert "owner" in create_resp.get_json()
|
||||
|
||||
# Revoke the certificate
|
||||
client.post(f"/pki/revoke/{serial}", headers={"X-SSL-Client-SHA1": issuer})
|
||||
|
||||
# Try to delete paste with revoked cert - should fail
|
||||
delete_resp = client.delete(f"/{paste_id}", headers={"X-SSL-Client-SHA1": cert_fingerprint})
|
||||
assert delete_resp.status_code == 401
|
||||
|
||||
|
||||
class TestPKICryptoFunctions:
|
||||
"""Test standalone PKI cryptographic functions."""
|
||||
|
||||
def test_derive_key_consistency(self):
|
||||
"""Key derivation produces consistent results."""
|
||||
from app.pki import derive_key
|
||||
|
||||
password = "test-password"
|
||||
salt = b"x" * 32
|
||||
|
||||
key1 = derive_key(password, salt)
|
||||
key2 = derive_key(password, salt)
|
||||
|
||||
assert key1 == key2
|
||||
assert len(key1) == 32
|
||||
|
||||
def test_encrypt_decrypt_roundtrip(self):
|
||||
"""Private key encryption/decryption roundtrip."""
|
||||
from cryptography.hazmat.primitives.asymmetric import ec
|
||||
|
||||
from app.pki import decrypt_private_key, encrypt_private_key
|
||||
|
||||
# Generate a test key
|
||||
private_key = ec.generate_private_key(ec.SECP384R1())
|
||||
password = "test-password"
|
||||
|
||||
# Encrypt
|
||||
encrypted, salt = encrypt_private_key(private_key, password)
|
||||
|
||||
# Decrypt
|
||||
decrypted = decrypt_private_key(encrypted, salt, password)
|
||||
|
||||
# Verify same key
|
||||
assert private_key.private_numbers() == decrypted.private_numbers()
|
||||
|
||||
def test_wrong_password_fails(self):
|
||||
"""Decryption with wrong password fails."""
|
||||
from cryptography.hazmat.primitives.asymmetric import ec
|
||||
|
||||
from app.pki import (
|
||||
InvalidPasswordError,
|
||||
decrypt_private_key,
|
||||
encrypt_private_key,
|
||||
)
|
||||
|
||||
private_key = ec.generate_private_key(ec.SECP384R1())
|
||||
encrypted, salt = encrypt_private_key(private_key, "correct")
|
||||
|
||||
with pytest.raises(InvalidPasswordError):
|
||||
decrypt_private_key(encrypted, salt, "wrong")
|
||||
|
||||
def test_fingerprint_calculation(self):
|
||||
"""Certificate fingerprint is calculated correctly."""
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
from cryptography import x509
|
||||
|
||||
# Minimal self-signed cert for testing
|
||||
from cryptography.hazmat.primitives import hashes
|
||||
from cryptography.hazmat.primitives.asymmetric import ec
|
||||
from cryptography.x509.oid import NameOID
|
||||
|
||||
from app.pki import calculate_fingerprint
|
||||
|
||||
key = ec.generate_private_key(ec.SECP256R1())
|
||||
subject = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "test")])
|
||||
cert = (
|
||||
x509.CertificateBuilder()
|
||||
.subject_name(subject)
|
||||
.issuer_name(subject)
|
||||
.public_key(key.public_key())
|
||||
.serial_number(1)
|
||||
.not_valid_before(datetime.now(UTC))
|
||||
.not_valid_after(datetime.now(UTC) + timedelta(days=1))
|
||||
.sign(key, hashes.SHA256())
|
||||
)
|
||||
|
||||
fingerprint = calculate_fingerprint(cert)
|
||||
|
||||
assert len(fingerprint) == 40
|
||||
assert all(c in "0123456789abcdef" for c in fingerprint)
|
||||
Reference in New Issue
Block a user