112 lines
3.7 KiB
Python
112 lines
3.7 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
"""
|
||
|
|
Test infrastructure for VeriFactu API certificate validation.
|
||
|
|
"""
|
||
|
|
|
||
|
|
import base64
|
||
|
|
import json
|
||
|
|
import os
|
||
|
|
import sys
|
||
|
|
from pathlib import Path
|
||
|
|
from urllib.request import urlopen, Request
|
||
|
|
from urllib.error import URLError
|
||
|
|
|
||
|
|
API_URL = "http://localhost:6789"
|
||
|
|
|
||
|
|
try:
|
||
|
|
from cryptography.hazmat.primitives import serialization
|
||
|
|
from cryptography.hazmat.primitives.asymmetric import padding
|
||
|
|
from cryptography.hazmat.backends import default_backend
|
||
|
|
HAS_CRYPTO = True
|
||
|
|
except ImportError:
|
||
|
|
HAS_CRYPTO = False
|
||
|
|
|
||
|
|
|
||
|
|
class VeriFactuTester:
|
||
|
|
def __init__(self):
|
||
|
|
self.api_url = API_URL
|
||
|
|
self.certs_dir = Path(__file__).parent / "certs"
|
||
|
|
|
||
|
|
def check_health(self):
|
||
|
|
"""Check if API is running."""
|
||
|
|
try:
|
||
|
|
req = Request(f"{self.api_url}/api/v1/health", method="GET")
|
||
|
|
with urlopen(req, timeout=5) as response:
|
||
|
|
return json.loads(response.read().decode())
|
||
|
|
except:
|
||
|
|
return None
|
||
|
|
|
||
|
|
def get_public_key(self):
|
||
|
|
"""Get public key from API."""
|
||
|
|
try:
|
||
|
|
req = Request(f"{self.api_url}/api/v1/auth/public-key", method="GET")
|
||
|
|
with urlopen(req, timeout=10) as response:
|
||
|
|
result = json.loads(response.read().decode())
|
||
|
|
return base64.b64decode(result["public_key"])
|
||
|
|
except Exception as e:
|
||
|
|
print(f"ERROR getting public key: {e}")
|
||
|
|
return None
|
||
|
|
|
||
|
|
def encrypt_password(self, public_key_pem, password):
|
||
|
|
"""Encrypt password with public key."""
|
||
|
|
if not HAS_CRYPTO:
|
||
|
|
print("WARNING: cryptography not available")
|
||
|
|
return base64.b64encode(password.encode()).decode()
|
||
|
|
|
||
|
|
try:
|
||
|
|
public_key = serialization.load_pem_public_key(public_key_pem, default_backend())
|
||
|
|
encrypted = public_key.encrypt(
|
||
|
|
password.encode(),
|
||
|
|
padding.PKCS1v15()
|
||
|
|
)
|
||
|
|
return base64.b64encode(encrypted).decode()
|
||
|
|
except Exception as e:
|
||
|
|
print(f"ERROR encrypting password: {e}")
|
||
|
|
return None
|
||
|
|
|
||
|
|
def register_certificate(self, cert_path, encrypted_password, cert_name):
|
||
|
|
"""Register certificate via API."""
|
||
|
|
data = {
|
||
|
|
"cert_name": cert_name,
|
||
|
|
"cert_path": cert_path,
|
||
|
|
"password_encrypted": encrypted_password
|
||
|
|
}
|
||
|
|
|
||
|
|
try:
|
||
|
|
req = Request(
|
||
|
|
f"{self.api_url}/api/v1/auth/register",
|
||
|
|
data=json.dumps(data).encode(),
|
||
|
|
method="POST"
|
||
|
|
)
|
||
|
|
req.add_header("Content-Type", "application/json")
|
||
|
|
with urlopen(req, timeout=30) as response:
|
||
|
|
return json.loads(response.read().decode())
|
||
|
|
except URLError as e:
|
||
|
|
return {"error": str(e), "success": False}
|
||
|
|
except Exception as e:
|
||
|
|
return {"error": str(e), "success": False}
|
||
|
|
|
||
|
|
def test_certificate(self, cert_file, password, expected_result, test_name):
|
||
|
|
"""Test a single certificate."""
|
||
|
|
print(f"\n--- Testing: {test_name} ---")
|
||
|
|
|
||
|
|
pub_key = self.get_public_key()
|
||
|
|
if not pub_key:
|
||
|
|
print("ERROR: Cannot get public key")
|
||
|
|
return False
|
||
|
|
|
||
|
|
enc_password = self.encrypt_password(pub_key, password)
|
||
|
|
if not enc_password:
|
||
|
|
print("ERROR: Cannot encrypt password")
|
||
|
|
return False
|
||
|
|
|
||
|
|
result = self.register_certificate(cert_file, enc_password, test_name)
|
||
|
|
|
||
|
|
print(f"API Response: {json.dumps(result, indent=2)}")
|
||
|
|
|
||
|
|
return result
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
print("This module should be imported, not run directly.")
|
||
|
|
print("Use: python test/run_tests.py")
|