Add unit tests to jwtproxy

.github/workflows/test-jwtproxy.yml

@@ -0,0 +1,34 @@
+name: Test jwtproxy
+permissions:
+  contents: read
+  pull-requests: write
+
+on:
+  push:
+    branches:
+    - features/**
+  pull_request:
+    branches:
+    - main
+    - master
+
+jobs:
+  docker:
+    timeout-minutes: 10
+    runs-on: ubuntu-latest
+    env:
+      INITIALIZE_DB: false
+
+    steps:
+    - name: Checkout
+      uses: actions/checkout@v1
+
+    - name: Start containers
+      run: docker compose -f "compose.yml" up proxy -d --build
+
+    - name: Run tests
+      run: docker compose exec -T proxy pytest -v --show-capture=no
+
+    - name: Stop containers
+      if: always()
+      run: docker compose -f "compose.yml" down

.pre-commit-config.yaml

@@ -48,3 +48,18 @@ repos:
       - id: trailing-whitespace
         args:
           - --markdown-linebreak-ext=md
+  - repo: https://github.com/astral-sh/ruff-pre-commit
+    rev: v0.12.11
+    hooks:
+      - id: ruff
+        files: ^jwtproxy/
+        args:
+          - --fix
+          - --show-fixes
+          - --config=jwtproxy/pyproject.toml
+          - --exit-non-zero-on-fix
+  - repo: https://github.com/psf/black
+    rev: 25.1.0
+    hooks:
+      - id: black
+        files: ^jwtproxy/

compose.yml

@@ -21,11 +21,9 @@ services:
       dockerfile: Dockerfile.dev
     ports:
       - 9191:8000
-    volumes:
-      - ./jwtproxy/src:/app/jwtproxy
     environment:
       PROXY_URL: http://map
-      JWKS_PATH: /app/jwtproxy/test_jwk.json
+      JWKS_PATH: /app/test_jwk.json
       LOG_LEVEL: DEBUG
       #JWKS_URL: "https://iam.amsterdam.nl/auth/realms/datapunt-ad-acc/protocol/openid-connect/certs"
 

jwtproxy/Dockerfile.dev

@@ -11,13 +11,14 @@ RUN apt-get update && apt-get install nginx -y
 COPY supervisord.conf /etc
 
 WORKDIR /app
-COPY src jwtproxy
+COPY src ./
 COPY nginx/nginx.conf /etc/nginx/nginx.conf
 COPY nginx/server.conf /etc/nginx/conf.d/server.conf
+COPY pyproject.toml .
 COPY requirements_dev.txt .
 COPY requirements.txt .
 
 RUN pip install supervisor
 RUN pip install -r requirements_dev.txt
 
-ENTRYPOINT [ "/bin/sh", "-c", "nginx -g 'daemon on;'; adev runserver /app/jwtproxy/server.py -p 8081 --app-factory main;"]
+ENTRYPOINT [ "/bin/sh", "-c", "nginx -g 'daemon on;'; adev runserver /app/server.py -p 8081 --app-factory main;"]

jwtproxy/Dockerfile.prd

@@ -3,7 +3,7 @@ FROM python:3.10-slim-bullseye
 COPY supervisord.conf /etc
 
 WORKDIR /app
-COPY src jwtproxy 
+COPY src ./
 COPY requirements.txt .
 
 RUN usermod --non-unique --uid 999 proxy

jwtproxy/manifest.yml

@@ -55,7 +55,7 @@ spec:
           - name: LOG_LEVEL
             value: DEBUG
           - name: JWKS_PATH
-            value: /app/jwtproxy/test_jwk.json
+            value: /app/test_jwk.json
 ---
 apiVersion: v1
 kind: Service

jwtproxy/pyproject.toml

@@ -0,0 +1,74 @@
+# ==== pytest ====
+[tool.pytest.ini_options]
+minversion = "6.0"
+asyncio_mode = "auto"
+testpaths = ["tests"]
+
+# ==== black ====
+[tool.black]
+line-length = 99
+target-version = ['py38']
+
+# ==== ruff ====
+[tool.ruff]
+line-length = 99
+target-version = "py38"
+
+[tool.ruff.lint]
+select = [
+    "F", # pyflakes
+    "E", # pycodestyle errors
+    "W", # pycodestyle warnings
+    "I", # isort
+    "B", # flake8-bugbear
+    "C90", # mccabe
+    "BLE", # flake8-blind-except
+    "C4", # flake8-comprehensions
+    "DTZ", # flake8-datetimez
+    "T10", # flake8-debugger
+    "DJ", # flake8-django
+    "ISC", # flake8-implicit-str-concat
+    "G", # flake8-logging-format
+    "PIE", # flake8-pie
+    "PGH", # pygrep-hooks
+    "RET", # flake8-return (partially)
+    # "PT",  # flake8-pytest-style
+    # "TCH",  # flake8-type-checking (moves import to `if typing.TYPE_CHECKING`)
+    # "ERA",  # eradicate (commented out code)
+    # "TRY",  # tryceratops
+    "SIM", # flake8-simplify
+    "TID", # flake8-tidy-imports
+    "INT", # flake8-gettext
+    "FLY", # flynt (fixes unneeded static string joins)
+    "UP", # pyupgrade
+    "S", # security (bugbear)
+    "RUF010", # ruff: fix f"{str(..)}" usage
+    "RUF013", # ruff: fix annotations for =None arguments
+]
+ignore = [
+    "S311", # allow random.randint()
+    "DJ001", # allow models.CharField(null=True)
+    "SIM105", # enforcing contextlib.suppress() instead of try..catch
+    "RET501", # unnecessary-return-none
+    "RET505", # superfluous-else-return
+    "RET505", # superfluous-else-return
+    "RET506", # superfluous-else-raise
+    "RET507", # superfluous-else-continue
+    "RET508", # superfluous-else-break
+    "S607", # subprocess partial path
+]
+
+[tool.ruff.lint.flake8-comprehensions]
+allow-dict-calls-with-keyword-arguments = true
+
+[tool.ruff.lint.flake8-gettext]
+extend-function-names = ["gettext_lazy", "ngettext_lazy", "pgettext", "pgettext_lazy", "npgettext", "npgettext_lazy"]
+
+[tool.ruff.lint.isort]
+known-first-party = ["tests"]
+
+[tool.ruff.lint.mccabe]
+max-complexity = 10
+
+[tool.ruff.lint.per-file-ignores]
+"tests/**/*.py" = ["DJ008", "S101", "S105", "S106", "S314", "S320", "S608"]  # allow asserts, hardcoded passwords, lxml parsing, SQL injection

jwtproxy/requirements_dev.txt

@@ -1,15 +1,20 @@
 -r requirements.txt
 
 aiohttp-devtools==1.0.post0
+aioresponses==0.7.8
 anyio==3.6.2
 asttokens==2.2.0
+black==25.11.0
 click==8.1.3
 devtools==0.10.0
 executing==1.2.0
+pre-commit==4.3.0
 Pygments==2.15.0
+pytest==8.4.2
+pytest-aiohttp==1.1.0
+ruff==0.14.4
 six==1.16.0
 sniffio==1.3.0
 watchgod==0.8.2
 wrapt==1.14.1
 yarl==1.17.0
-

jwtproxy/src/jwt_keygen.py

@@ -20,28 +20,23 @@
 """
 
 parser = argparse.ArgumentParser()
-parser.add_argument(
-    "scopes", nargs="*", help="Scopes added to the JWT as private claims"
-)
-parser.add_argument(
-    "--exp", default=None, type=int, help="Number of seconds to expiration"
-)
+parser.add_argument("scopes", nargs="*", help="Scopes added to the JWT as private claims")
+parser.add_argument("--exp", default=None, type=int, help="Number of seconds to expiration")
 
 JWK_PATH = pathlib.Path(__file__).parent / "test_jwk.json"
 
 
 def generate_jwk():
-    return JWK.generate(
-        kty="EC", crv="P-256", kid=str(uuid.uuid4()), key_ops=["verify", "sign"]
-    )
+    return JWK.generate(kty="EC", crv="P-256", kid=str(uuid.uuid4()), key_ops=["verify", "sign"])
 
 
 def generate_jwt(scopes, exp):
     """Generate a JWT, signed using the first key stored at test_jwk.json.
     If there is no JWKSet at this location, generate it and write it to
     the file as a JWKSet"""
     if JWK_PATH.exists():
-        key = JWK(**json.load(open(JWK_PATH))["keys"][0])
+        with open(JWK_PATH) as f:
+            key = JWK(**json.load(f)["keys"][0])
         if "d" not in key:
             raise ValueError("The configured JWK does not contain a private key")
     else:
@@ -53,9 +48,7 @@ def generate_jwt(scopes, exp):
     now = int(time.time())
     claims = {
         "iat": now,
-        "realm_access": {
-            "roles": scopes
-        },
+        "realm_access": {"roles": scopes},
         "sub": "test@tester.nl",
     }
     if exp is not None:
@@ -64,10 +57,11 @@ def generate_jwt(scopes, exp):
     token = JWT(header={"alg": "ES256", "kid": key.kid, "kty": "EC"}, claims=claims)
     token.make_signed_token(key)
 
-    sys.stdout.write(token.serialize())
-    sys.stdout.write("\n")
+    return token.serialize()
 
 
 if __name__ == "__main__":
     args = parser.parse_args()
-    generate_jwt(args.scopes, args.exp)
+    token = generate_jwt(args.scopes, args.exp)
+    sys.stdout.write(token)
+    sys.stdout.write("\n")

jwtproxy/src/server.py

@@ -4,15 +4,14 @@
 import asyncio
 import json
 import logging
+import re
 from logging.config import dictConfig
 from os import environ as env
-import re
 
 from aiohttp import ClientSession, ClientTimeout, DummyCookieJar, TCPConnector, web
-from jwcrypto.jwt import JWT, JWKSet
+from jwcrypto.jwt import JWT, JWKSet, JWTExpired
 from yarl import URL
 
-
 LOG_LEVEL = env.get("LOG_LEVEL", "INFO")
 CHUNK_SIZE = 1024
 CONN_POOL_SIZE = 100
@@ -91,6 +90,7 @@
 audit_logger = logging.getLogger("jwtproxy.auditlogs")
 access_logger = logging.getLogger("jwtproxy.accesslogs")
 
+
 async def fetch_token(req: web.Request):
     """Fetch the token from the request.
 
@@ -124,22 +124,26 @@ async def audit(req: web.Request, token: str):
 
     try:
         j.deserialize(token, key=jwks)
-    except Exception as e:
+    except ValueError as e:
         app_logger.warning(e)
-        raise web.HTTPForbidden(reason="Invalid token")
+        raise web.HTTPUnauthorized(reason="Invalid token") from e
+    except JWTExpired as e:
+        app_logger.warning(e)
+        raise web.HTTPUnauthorized(reason="Token expired") from e
 
     claims = json.loads(j.claims)
     try:
-        assert "fp_mdw" in claims["realm_access"]["roles"]
-    except (KeyError, AssertionError):
+        if "fp_mdw" not in claims["realm_access"]["roles"]:
+            raise ValueError("Required scope not present in claims")
+    except (KeyError, ValueError):
         audit_logger.info(
             "Access to %s (%s) denied for subject %s (username: %s)",
             req.url,
             req.method,
             claims.get("email"),
             claims.get("preferred_username"),
         )
-        raise web.HTTPUnauthorized(reason="Insufficient access privilege")
+        raise web.HTTPForbidden(reason="Insufficient access privilege") from None
 
     audit_logger.info(
         "Access to %s (%s) granted to subject %s (username: %s)",
@@ -203,7 +207,8 @@ async def get_jwk():
             ) as resp:
                 payload = await resp.text()
         else:
-            payload = open(env.get("JWKS_PATH")).read()
+            with open(env.get("JWKS_PATH")) as file:
+                payload = file.read()
         cached_jwk = JWKSet.from_json(payload)
         return cached_jwk
 
@@ -214,7 +219,10 @@ async def check_protection(path, method):
         if method == "OPTIONS":
             return web.Response(text="Status OK")
         else:
-            raise web.HTTPForbidden(reason=f"Accessing `{path}` without a JWT token is not allowed.")
+            raise web.HTTPForbidden(
+                reason=f"Accessing `{path}` without a JWT token is not allowed."
+            )
+    return None
 
 
 async def handle(req: web.Request):
@@ -230,15 +238,13 @@ async def handle(req: web.Request):
     app_logger.debug("Request params: \n %s", req.rel_url.query)
     app_logger.debug("Request headers: \n %s", req.headers)
 
-
     if token is not None:
         await audit(req, token)
 
     body = None
     if req.can_read_body:
         body = await req.text()
         app_logger.debug("Request body: \n %s", body)
-
     async with SessionManager.session().request(
         req.method, target_url, headers=req.headers, params=req.rel_url.query, data=body
     ) as resp:
@@ -267,7 +273,7 @@ async def health(req: web.Request):
 parser.add_argument("--port", default=8080, type=int)
 
 
-async def main(*argv):
+async def main(*argv) -> web.Application:
     app = web.Application()
     app.router.add_route("GET", "/status/health", health)
     app.router.add_route("OPTIONS", "/{path:.*?}", handle)
@@ -283,6 +289,6 @@ async def main(*argv):
 
     # support listening on unix domain sockets and ports
     if args.path:
-        web.run_app(main(), host="0.0.0.0", path=args.path, access_log=access_logger)
+        web.run_app(main(), host="0.0.0.0", path=args.path, access_log=access_logger)  # noqa: S104
     else:
-        web.run_app(main(), host="0.0.0.0", port=args.port, access_log=access_logger)
+        web.run_app(main(), host="0.0.0.0", port=args.port, access_log=access_logger)  # noqa: S104