aboutsummaryrefslogtreecommitdiff
path: root/gn_auth
diff options
context:
space:
mode:
authorFrederick Muriuki Muriithi2024-07-18 12:17:37 -0500
committerFrederick Muriuki Muriithi2024-07-18 12:21:42 -0500
commiteffe7065eba87579551d86f306eba830f9cc8d6f (patch)
treeca8936aa76ee03c6864f156cf6f3273a258982ff /gn_auth
parent2e6f8a5d6bde0bd8fdfb9217b352ea2914ba592f (diff)
downloadgn-auth-effe7065eba87579551d86f306eba830f9cc8d6f.tar.gz
Add module to handle JWK files
Diffstat (limited to 'gn_auth')
-rw-r--r--gn_auth/auth/jwks.py67
1 files changed, 67 insertions, 0 deletions
diff --git a/gn_auth/auth/jwks.py b/gn_auth/auth/jwks.py
new file mode 100644
index 0000000..7c091b8
--- /dev/null
+++ b/gn_auth/auth/jwks.py
@@ -0,0 +1,67 @@
+"""Utilities dealing with JSON Web Keys (JWK)"""
+import os
+from pathlib import Path
+from typing import Any, Union
+from datetime import datetime
+
+from authlib.jose import JsonWebKey
+from pymonad.either import Left, Right, Either
+
+def jwks_directory(appsecretsdir: Path) -> Path:
+ """Compute the directory where the JWKs are stored."""
+ if appsecretsdir.exists() and appsecretsdir.is_dir():
+ jwksdir = Path(appsecretsdir, "jwks/")
+ if not jwksdir.exists():
+ jwksdir.mkdir()
+ return jwksdir
+ raise ValueError(
+ "The `appsecretsdir` value should be a directory that actually exists.")
+
+
+def generate_and_save_private_key(
+ storagedir: Path,
+ kty: str = "RSA",
+ crv_or_size: Union[str, int] = 2048,
+ options: tuple[tuple[str, Any]] = (("iat", datetime.now().timestamp()),)
+) -> JsonWebKey:
+ """Generate a private key and save to `storagedir`."""
+ privatejwk = JsonWebKey.generate_key(
+ kty, crv_or_size, dict(options), is_private=True)
+ keyname = f"{privatejwk.thumbprint()}.private.pem"
+ with open(Path(storagedir, keyname), "wb") as pemfile:
+ pemfile.write(privatejwk.as_pem(is_private=True))
+
+ return privatejwk
+
+
+def pem_to_jwk(filepath: Path) -> JsonWebKey:
+ """Parse a PEM file into a JWK object."""
+ with open(filepath, "rb") as pemfile:
+ return JsonWebKey.import_key(pemfile.read())
+
+
+def __sorted_jwks_paths__(storagedir: Path) -> tuple[tuple[float, Path], ...]:
+ """A sorted list of the JWK file paths with their creation timestamps."""
+ return tuple(sorted(((os.stat(keypath).st_ctime, keypath)
+ for keypath in (Path(storagedir, keyfile)
+ for keyfile in os.listdir(storagedir)
+ if keyfile.endswith(".pem"))),
+ key=lambda tpl: tpl[0]))
+
+
+def list_jwks(storagedir: Path) -> tuple[JsonWebKey, ...]:
+ """
+ List all the JWKs in a particular directory in the order they were created.
+ """
+ return tuple(pem_to_jwk(keypath) for ctime,keypath in
+ __sorted_jwks_paths__(storagedir))
+
+
+def newest_jwk(storagedir: Path) -> Either:
+ """
+ Return an Either monad with the newest JWK or a message if none exists.
+ """
+ existingkeys = __sorted_jwks_paths__(storagedir)
+ if len(existingkeys) > 0:
+ return Right(pem_to_jwk(existingkeys[-1][1]))
+ return Left("No JWKs exist")