Merge pull request #110 from dnstapir/use_dnstapir_module

Use shared Python module

.pre-commit-config.yaml

@@ -11,7 +11,7 @@ repos:
       - id: check-yaml
   - repo: https://github.com/astral-sh/ruff-pre-commit
     # Ruff version.
-    rev: v0.6.5
+    rev: v0.7.2
     hooks:
       # Run the linter.
       - id: ruff

aggrec/__init__.py

@@ -3,7 +3,7 @@
 __version__ = version("aggrec")
 
 try:
-    from aggrec.buildinfo import __commit__, __timestamp__
+    from .buildinfo import __commit__, __timestamp__
 
     __verbose_version__ = f"{__version__} ({__commit__})"
 except ModuleNotFoundError:

aggrec/aggregates.py

@@ -238,9 +238,7 @@ async def create_aggregate(
     span = trace.get_current_span()
 
     with tracer.start_as_current_span("http_request_verifier"):
-        http_request_verifier = RequestVerifier(
-            client_database=request.app.settings.clients_database, key_cache=request.app.key_cache
-        )
+        http_request_verifier = RequestVerifier(key_resolver=request.app.key_resolver)
         res = await http_request_verifier.verify(request)
 
     creator = res.parameters.get("keyid")

aggrec/helpers.py

@@ -14,10 +14,8 @@
 )
 from http_message_signatures.algorithms import signature_algorithms as supported_signature_algorithms
 from http_message_signatures.exceptions import InvalidSignature
-from pydantic import AnyHttpUrl, DirectoryPath
 
-from .key_cache import KeyCache
-from .key_resolver import FileKeyResolver, UrlKeyResolver
+from dnstapir.key_resolver import KeyResolver, PublicKey
 
 DEFAULT_SIGNATURE_ALGORITHM = algorithms.ECDSA_P256_SHA256
 HASH_ALGORITHMS = {"sha-256": hashlib.sha256, "sha-512": hashlib.sha512}
@@ -39,25 +37,22 @@ class ContentDigestMissing(ContentDigestException):
     pass
 
 
+class CustomHTTPSignatureKeyResolver(HTTPSignatureKeyResolver):
+    def __init__(self, key_resolver: KeyResolver):
+        self.key_resolver = key_resolver
+
+    def resolve_public_key(self, key_id: str) -> PublicKey:
+        return self.key_resolver.resolve_public_key(key_id=key_id)
+
+
 class RequestVerifier:
     def __init__(
         self,
+        key_resolver: KeyResolver,
         algorithm: HTTPSignatureAlgorithm | None = None,
-        key_resolver: HTTPSignatureKeyResolver | None = None,
-        client_database: AnyHttpUrl | DirectoryPath | None = None,
-        key_cache: KeyCache | None = None,
     ):
         self.algorithm = algorithm or DEFAULT_SIGNATURE_ALGORITHM
-        if key_resolver:
-            self.key_resolver = key_resolver
-        elif client_database and (
-            str(client_database).startswith("http://") or str(client_database).startswith("https://")
-        ):
-            self.key_resolver = UrlKeyResolver(client_database_base_url=str(client_database), key_cache=key_cache)
-        elif client_database:
-            self.key_resolver = FileKeyResolver(client_database_directory=str(client_database), key_cache=key_cache)
-        else:
-            raise ValueError("No key resolver nor client database specified")
+        self.http_key_resolver = CustomHTTPSignatureKeyResolver(key_resolver)
         self.logger = logging.getLogger(__name__).getChild(self.__class__.__name__)
 
     async def verify_content_digest(self, result: VerifyResult, request: Request):
@@ -87,7 +82,7 @@ async def verify(self, request: Request) -> VerifyResult:
         signature_algorithm = supported_signature_algorithms[alg]
         verifier = HTTPMessageVerifier(
             signature_algorithm=signature_algorithm,
-            key_resolver=self.key_resolver,
+            key_resolver=self.http_key_resolver,
         )
         try:
             results = verifier.verify(request)

aggrec/server.py

@@ -5,55 +5,22 @@
 import aiomqtt
 import boto3
 import mongoengine
-import redis
 import uvicorn
 from fastapi import FastAPI
 from uvicorn.middleware.proxy_headers import ProxyHeadersMiddleware
 
 import aggrec.aggregates
 import aggrec.extras
+from dnstapir.key_cache import key_cache_from_settings
+from dnstapir.key_resolver import key_resolver_from_client_database
+from dnstapir.logging import configure_json_logging
+from dnstapir.opentelemetry import configure_opentelemetry
 
 from . import OPENAPI_METADATA, __verbose_version__
-from .key_cache import CombinedKeyCache, KeyCache, MemoryKeyCache, RedisKeyCache
-from .logging import JsonFormatter  # noqa
 from .settings import Settings
-from .telemetry import configure_opentelemetry
 
 logger = logging.getLogger(__name__)
 
-LOGGING_RECORD_CUSTOM_FORMAT = {
-    "time": "asctime",
-    # "Created": "created",
-    # "RelativeCreated": "relativeCreated",
-    "name": "name",
-    # "Levelno": "levelno",
-    "levelname": "levelname",
-    "process": "process",
-    "thread": "thread",
-    # "threadName": "threadName",
-    # "Pathname": "pathname",
-    # "Filename": "filename",
-    # "Module": "module",
-    # "Lineno": "lineno",
-    # "FuncName": "funcName",
-    "message": "message",
-}
-
-LOGGING_CONFIG_JSON = {
-    "version": 1,
-    "disable_existing_loggers": False,
-    "formatters": {
-        "json": {
-            "class": "aggrec.logging.JsonFormatter",
-            "format": LOGGING_RECORD_CUSTOM_FORMAT,
-        },
-    },
-    "handlers": {
-        "json": {"class": "logging.StreamHandler", "formatter": "json"},
-    },
-    "root": {"handlers": ["json"], "level": "DEBUG"},
-}
-
 
 class AggrecServer(FastAPI):
     def __init__(self, settings: Settings):
@@ -63,26 +30,18 @@ def __init__(self, settings: Settings):
         self.add_middleware(ProxyHeadersMiddleware)
         self.include_router(aggrec.aggregates.router)
         self.include_router(aggrec.extras.router)
-        configure_opentelemetry(
-            self,
-            service_name="aggrec",
-            spans_endpoint=str(settings.otlp.spans_endpoint),
-            metrics_endpoint=str(settings.otlp.metrics_endpoint),
-            insecure=settings.otlp.insecure,
+        if self.settings.otlp:
+            configure_opentelemetry(
+                service_name="aggrec",
+                settings=self.settings.otlp,
+                fastapi_app=self,
+            )
+        else:
+            self.logger.info("Configured without OpenTelemetry")
+        key_cache = key_cache_from_settings(self.settings.key_cache) if self.settings.key_cache else None
+        self.key_resolver = key_resolver_from_client_database(
+            client_database=str(self.settings.clients_database), key_cache=key_cache
         )
-        self.key_cache: KeyCache | None = None
-        if self.settings.key_cache:
-            memory_key_cache = MemoryKeyCache(size=self.settings.key_cache.size, ttl=self.settings.key_cache.ttl)
-            if redis_settings := self.settings.key_cache.redis:
-                redis_client = redis.StrictRedis(host=redis_settings.host, port=redis_settings.port)
-                self.logger.debug("Using REDIS at %s:%d", redis_settings.host, redis_settings.port)
-                redis_key_cache = RedisKeyCache(redis_client=redis_client, ttl=self.settings.key_cache.ttl)
-                if self.settings.key_cache.size:
-                    self.key_cache = CombinedKeyCache([memory_key_cache, redis_key_cache])
-                else:
-                    self.key_cache = redis_key_cache
-            elif self.settings.key_cache.size:
-                self.key_cache = memory_key_cache
 
     @staticmethod
     def connect_mongodb(settings: Settings):
@@ -143,8 +102,7 @@ def main() -> None:
         print(f"Aggregate Receiver version {__verbose_version__}")
         return
 
-    logging_config = LOGGING_CONFIG_JSON
-    logging.config.dictConfig(logging_config)
+    logging_config = configure_json_logging()
 
     if args.debug:
         logging.basicConfig(level=logging.DEBUG)