[codex] Use Cloudflare Access for Dragonfly API

src/bot/__main__.py

@@ -30,13 +30,9 @@ async def main() -> None:
     async with ClientSession(headers={"Content-Type": "application/json"}, timeout=ClientTimeout(total=30)) as session:
         dragonfly_services = DragonflyServices(
             session=session,
-            base_url=constants.Dragonfly.base_url,
-            auth_url=constants.Dragonfly.auth_url,
-            audience=constants.Dragonfly.audience,
+            base_url=constants.DragonflyConfig.api_url,
             client_id=constants.Dragonfly.client_id,
             client_secret=constants.Dragonfly.client_secret,
-            username=constants.Dragonfly.username,
-            password=constants.Dragonfly.password,
         )
 
         bot = Bot(

src/bot/constants.py

@@ -57,16 +57,11 @@ class _ThreatIntelFeed(EnvConfig, env_prefix="tif_"):
 DEBUG_MODE = Miscellaneous.debug
 
 
-class _Dragonfly(EnvConfig, env_prefix="auth0_"):
-    """Configuration for the Dragonfly API."""
+class _Dragonfly(EnvConfig, env_prefix="cf_access_"):
+    """Cloudflare Access configuration for the Dragonfly API."""
 
-    base_url: str = "https://dragonfly.vipyrsec.com"
-    auth_url: str = "https://vipyrsec.us.auth0.com/oauth/token"
-    audience: str = "https://dragonfly.vipyrsec.com"
     client_id: str = ""
     client_secret: str = ""
-    username: str = ""
-    password: str = ""
 
 
 Dragonfly = _Dragonfly()

src/bot/dragonfly_services.py

@@ -2,7 +2,7 @@
 
 import dataclasses
 from dataclasses import dataclass
-from datetime import UTC, datetime, timedelta
+from datetime import datetime
 from enum import Enum
 from typing import Any, Self
 
@@ -60,47 +60,25 @@ class PackageReport:
 class DragonflyServices:
     """A class wrapping Dragonfly's API."""
 
-    def __init__(  # noqa: PLR0913 -- Maybe pass the entire constants class?
+    def __init__(
         self: Self,
         session: ClientSession,
         base_url: str,
-        auth_url: str,
-        audience: str,
         client_id: str,
         client_secret: str,
-        username: str,
-        password: str,
     ) -> None:
         """Initialize the DragonflyServices class."""
         self.session = session
         self.base_url = base_url
-        self.auth_url = auth_url
-        self.audience = audience
         self.client_id = client_id
         self.client_secret = client_secret
-        self.username = username
-        self.password = password
-        self.token = ""
-        self.token_expires_at = datetime.now(tz=UTC)
-
-    async def _update_token(self: Self) -> None:
-        """Update the OAUTH token."""
-        if self.token_expires_at > datetime.now(tz=UTC):
-            return
-
-        auth_dict = {
-            "grant_type": "password",
-            "audience": self.audience,
-            "client_id": self.client_id,
-            "client_secret": self.client_secret,
-            "username": self.username,
-            "password": self.password,
+
+    def _build_access_headers(self: Self) -> dict[str, str]:
+        """Build Cloudflare Access service-token headers."""
+        return {
+            "CF-Access-Client-Id": self.client_id,
+            "CF-Access-Client-Secret": self.client_secret,
         }
-        async with self.session.post(self.auth_url, json=auth_dict) as response:
-            response.raise_for_status()
-            data = await response.json()
-            self.token = data["access_token"]
-            self.token_expires_at = datetime.now(tz=UTC) + timedelta(seconds=data["expires_in"])
 
     async def make_request(
         self: Self,
@@ -110,14 +88,10 @@ async def make_request(
         json: dict[str, Any] | None = None,
     ) -> dict:  # type: ignore[type-arg]
         """Make a request to Dragonfly's API."""
-        await self._update_token()
-
-        headers = {"Authorization": "Bearer " + self.token}
-
         args = {
             "url": self.base_url + path,
             "method": method,
-            "headers": headers,
+            "headers": self._build_access_headers(),
         }
 
         if params is not None:

tests/test_dragonfly_services.py

@@ -0,0 +1,50 @@
+"""Tests for the Dragonfly API wrapper."""
+
+from __future__ import annotations
+
+import asyncio
+from typing import Any, Self
+from unittest.mock import Mock
+
+from bot.dragonfly_services import DragonflyServices
+
+
+class _MockResponse:
+    def __init__(self, payload: dict[str, Any]) -> None:
+        self._payload = payload
+        self.raise_for_status = Mock()
+
+    async def __aenter__(self) -> Self:
+        return self
+
+    async def __aexit__(self, exc_type: object, exc: object, tb: object) -> None:
+        return None
+
+    async def json(self) -> dict[str, Any]:
+        return self._payload
+
+
+def test_make_request_uses_cf_access_headers() -> None:
+    response = _MockResponse({"ok": True})
+    session = Mock()
+    session.request.return_value = response
+    service = DragonflyServices(
+        session=session,
+        base_url="https://dragonfly-staging.vipyrsec.com",
+        client_id="client-id",
+        client_secret="client-secret",
+    )
+
+    payload = asyncio.run(service.make_request("GET", "/package", params={"since": 1}))
+
+    assert payload == {"ok": True}
+    session.request.assert_called_once_with(
+        url="https://dragonfly-staging.vipyrsec.com/package",
+        method="GET",
+        headers={
+            "CF-Access-Client-Id": "client-id",
+            "CF-Access-Client-Secret": "client-secret",
+        },
+        params={"since": 1},
+    )
+    response.raise_for_status.assert_called_once_with()