Allow to persist the created CA

Author: julien-duponchelle

.gitignore

@@ -4,6 +4,7 @@ uv.lock
 .pytest_cache
 .ruff_cache
 docs/_*
+*.pem
 
 .serena
 CLAUDE.md

docs/custom-ca.md

@@ -17,6 +17,80 @@ To intercept HTTPS traffic, you need to create your own CA certificate and confi
 to trust this CA. This allows the proxy to generate and sign certificates for the target domains on-the-fly,
 allowing it to decrypt and inspect the HTTPS traffic.
 
+By default the library create a temporary Root CA on the fly but if you want to trust the CA in your
+browser you will need the persist the CA.
+
+## Using Custom CA Keys with TLSStore
+
+The `TLSStore` class can accept an existing CA private key and certificate instead of generating new ones. This is useful when you want to:
+
+- Reuse the same CA across multiple proxy runs
+- Use a pre-existing CA certificate
+- Maintain certificate consistency
+
+### Basic Usage
+
+```python
+from asyncio_https_proxy import TLSStore
+
+# Load existing CA key and certificate from files using the proper method
+tls_store = TLSStore.load_ca_from_disk("ca_private_key.pem", "ca_certificate.pem")
+```
+
+### Persisting CA to Disk
+
+To persist a CA to disk for reuse across multiple proxy runs:
+
+```python
+from asyncio_https_proxy import TLSStore
+
+# Create a new TLS store (will generate a new CA)
+tls_store = TLSStore()
+
+# Save the CA to disk for future use
+tls_store.save_ca_to_disk("ca_private_key.pem", "ca_certificate.pem")
+```
+
+### Complete Persistence Example
+
+```python
+from asyncio_https_proxy import TLSStore
+import os
+
+# Check if CA files already exist
+if os.path.exists("ca_private_key.pem") and os.path.exists("ca_certificate.pem"):
+    # Load existing CA from disk
+    tls_store = TLSStore.load_ca_from_disk("ca_private_key.pem", "ca_certificate.pem")
+    print("Loaded existing CA from disk")
+else:
+    # Create new CA and save it to disk
+    tls_store = TLSStore()
+    tls_store.save_ca_to_disk("ca_private_key.pem", "ca_certificate.pem")
+    print("Created new CA and saved to disk")
+```
+
+### Persistent CA Example
+
+For a complete example of creating and reusing a CA across multiple proxy runs, see [`examples/persistent_ca_usage.py`](../examples/persistent_ca_usage.py). This example demonstrates:
+
+- Generating a CA and saving it to disk if it doesn't exist
+- Loading an existing CA from disk on subsequent runs
+- Proper error handling for CA file operations
+
+```python
+# Run the persistent CA example
+python examples/persistent_ca_usage.py
+```
+
+The example will create `ca_private_key.pem` and `ca_certificate.pem` files in the current directory and reuse them on subsequent runs.
+
+### Important Considerations
+
+- Both `ca_key` and `ca_cert` parameters must be provided together or neither
+- The CA key must be an EllipticCurve private key (SECP256R1)
+- Store CA private keys securely and restrict file permissions appropriately
+- Consider the certificate validity period when reusing CA certificates
+
 ## Usage of custom CA in your applications
 
 ### Curl

examples/persistent_ca_usage.py

@@ -0,0 +1,94 @@
+#!/usr/bin/env python3
+"""
+Persistent CA usage example for asyncio-https-proxy.
+
+This example demonstrates how to create and reuse a Certificate Authority (CA) 
+across multiple proxy runs. If CA files don't exist, creates a default TLSStore 
+and saves its CA to disk for reuse on subsequent runs.
+"""
+
+import asyncio
+from pathlib import Path
+
+from asyncio_https_proxy import HTTPSForwardProxyHandler, TLSStore, start_proxy_server
+
+CA_KEY_FILE = "ca_private_key.pem"
+CA_CERT_FILE = "ca_certificate.pem"
+
+
+def get_or_create_ca():
+    """Get existing CA from disk or create a new TLSStore and persist its CA."""
+    ca_files_exist = Path(CA_KEY_FILE).exists() and Path(CA_CERT_FILE).exists()
+    
+    if ca_files_exist:
+        print("Loading existing CA from disk...")
+        tls_store = TLSStore.load_ca_from_disk(CA_KEY_FILE, CA_CERT_FILE)
+        print("✅ CA loaded from disk")
+        return tls_store
+    else:
+        print("No existing CA files found.")
+    
+    # Create new TLSStore with default CA generation
+    print("Creating new TLS store...")
+    tls_store = TLSStore()
+    
+    # Save the generated CA to disk for future use
+    print("Saving CA to disk for future reuse...")
+    tls_store.save_ca_to_disk(CA_KEY_FILE, CA_CERT_FILE)
+    print(f"✅ CA key saved to: {CA_KEY_FILE}")
+    print(f"✅ CA certificate saved to: {CA_CERT_FILE}")
+    
+    return tls_store
+
+
+class LoggingForwardProxyHandler(HTTPSForwardProxyHandler):
+    """Example forward proxy handler with logging."""
+    
+    async def on_client_connected(self):
+        print(f"Client connected: {self.request}")
+        await super().on_client_connected()
+
+    async def on_request_received(self):
+        print(f"Request: {self.request.method} {self.request.url()}")
+        await super().on_request_received()
+
+
+async def main():
+    """Run a proxy with persistent CA."""
+    
+    host = "127.0.0.1"
+    port = 8888
+    
+    print("=" * 60)
+    print("HTTPS Forward Proxy with Persistent CA")
+    print("=" * 60)
+    
+    # Get existing CA from disk or create new TLSStore and persist its CA
+    tls_store = get_or_create_ca()
+    
+    print(f"\nStarting HTTPS forward proxy on {host}:{port}")
+    print("\nTest the proxy with:")
+    print(f"  curl --cacert {CA_CERT_FILE} --proxy http://{host}:{port} https://httpbin.org/get")
+    print(f"  curl --cacert {CA_CERT_FILE} --proxy http://{host}:{port} http://httpbin.org/get")
+    print("\nPress Ctrl+C to stop the proxy")
+    print("=" * 60)
+
+    server = await start_proxy_server(
+        handler_builder=lambda: LoggingForwardProxyHandler(),
+        host=host,
+        port=port,
+        tls_store=tls_store,
+    )
+    
+    async with server:
+        try:
+            await server.serve_forever()
+        except KeyboardInterrupt:
+            print("\nShutting down proxy...")
+            server.close()
+            await server.wait_closed()
+            print("Proxy shut down.")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

src/asyncio_https_proxy/tls_store.py

@@ -1,6 +1,8 @@
 import datetime
 import ssl
 import tempfile
+from pathlib import Path
+from typing import Optional, Union
 
 from cryptography import x509
 from cryptography.hazmat.primitives import hashes, serialization
@@ -13,10 +15,25 @@
 class TLSStore:
     """
     A simple in-memory TLS store that generates a CA and signs certificates for domains on the fly.
+    
+    Args:
+        ca_key: Optional CA private key. If provided, ca_cert must also be provided.
+        ca_cert: Optional CA certificate. If provided, ca_key must also be provided.        
     """
 
-    def __init__(self):
-        self._ca = self._generate_ca()
+    def __init__(
+        self,
+        ca_key: Optional[ec.EllipticCurvePrivateKey] = None,
+        ca_cert: Optional[x509.Certificate] = None,
+    ):
+        if (ca_key is None) != (ca_cert is None):
+            raise ValueError("Both ca_key and ca_cert must be provided together, or neither")
+        
+        if ca_key is not None and ca_cert is not None:
+            self._ca = (ca_key, ca_cert)
+        else:
+            self._ca = self._generate_ca()
+        
         self._store = {}
 
     def _generate_ca(self):
@@ -181,3 +198,60 @@ def get_ssl_context(self, domain: str) -> ssl.SSLContext:
             ssl_context.load_cert_chain(certfile=cert_file.name, keyfile=key_file.name)
 
         return ssl_context
+
+    def save_ca_to_disk(self, key_file: Union[str, Path], cert_file: Union[str, Path]) -> None:
+        """
+        Save the CA private key and certificate to disk files.
+        
+        Args:
+            key_file: Path where to save the CA private key (PEM format)
+            cert_file: Path where to save the CA certificate (PEM format)
+        """
+        ca_key, ca_cert = self._ca
+        
+        # Save private key to disk
+        with open(key_file, "wb") as f:
+            f.write(ca_key.private_bytes(
+                encoding=serialization.Encoding.PEM,
+                format=serialization.PrivateFormat.PKCS8,
+                encryption_algorithm=serialization.NoEncryption(),
+            ))
+        
+        # Save certificate to disk
+        with open(cert_file, "wb") as f:
+            f.write(ca_cert.public_bytes(serialization.Encoding.PEM))
+
+    @classmethod
+    def load_ca_from_disk(
+        cls, 
+        key_file: Union[str, Path], 
+        cert_file: Union[str, Path]
+    ) -> "TLSStore":
+        """
+        Load CA private key and certificate from disk and create a TLSStore instance.
+        
+        Args:
+            key_file: Path to the CA private key file (PEM format)
+            cert_file: Path to the CA certificate file (PEM format)
+            
+        Returns:
+            TLSStore instance using the loaded CA
+            
+        Raises:
+            ValueError: If the key file doesn't contain an EllipticCurve private key
+            FileNotFoundError: If either file doesn't exist
+            ValueError: If the files cannot be parsed
+        """
+        # Load private key
+        with open(key_file, "rb") as f:
+            ca_key = serialization.load_pem_private_key(f.read(), password=None)
+        
+        # Verify it's an EC key
+        if not isinstance(ca_key, ec.EllipticCurvePrivateKey):
+            raise ValueError(f"CA key must be an EllipticCurve private key, got {type(ca_key)}")
+        
+        # Load certificate
+        with open(cert_file, "rb") as f:
+            ca_cert = x509.load_pem_x509_certificate(f.read())
+        
+        return cls(ca_key=ca_key, ca_cert=ca_cert)