providers/oauth2: more x5c and ecdsa x/y tests (#4463)

* add option to exclude x5*

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

#4082

* cleanup jwks, add flaky test

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

* add workaround based on https://github.com/jpadilla/pyjwt/issues/709

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

* don't rstrip hashes

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

* keycloak seems to strip equals

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

Signed-off-by: Jens Langhammer <jens@goauthentik.io>
This commit is contained in:
Jens L 2023-01-18 19:11:36 +01:00 committed by GitHub
parent f09305a444
commit e390f5b2d1
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 100 additions and 27 deletions

View File

@ -1,14 +1,42 @@
"""JWKS tests""" """JWKS tests"""
import base64
import json import json
from cryptography.hazmat.backends import default_backend
from cryptography.x509 import load_der_x509_certificate
from django.urls.base import reverse from django.urls.base import reverse
from jwt import PyJWKSet from jwt import PyJWKSet
from authentik.core.models import Application from authentik.core.models import Application
from authentik.core.tests.utils import create_test_cert, create_test_flow from authentik.core.tests.utils import create_test_cert, create_test_flow
from authentik.crypto.models import CertificateKeyPair
from authentik.lib.generators import generate_id
from authentik.providers.oauth2.models import OAuth2Provider from authentik.providers.oauth2.models import OAuth2Provider
from authentik.providers.oauth2.tests.utils import OAuthTestCase from authentik.providers.oauth2.tests.utils import OAuthTestCase
TEST_CORDS_CERT = """
-----BEGIN CERTIFICATE-----
MIIB6jCCAZCgAwIBAgIRAOsdE3N7zETzs+7shTXGj5wwCgYIKoZIzj0EAwIwHjEc
MBoGA1UEAwwTYXV0aGVudGlrIDIwMjIuMTIuMjAeFw0yMzAxMTYyMjU2MjVaFw0y
NDAxMTIyMjU2MjVaMHgxTDBKBgNVBAMMQ0NsbDR2TzFJSGxvdFFhTGwwMHpES2tM
WENYdzRPUFF2eEtZN1NrczAuc2VsZi1zaWduZWQuZ29hdXRoZW50aWsuaW8xEjAQ
BgNVBAoMCWF1dGhlbnRpazEUMBIGA1UECwwLU2VsZi1zaWduZWQwWTATBgcqhkjO
PQIBBggqhkjOPQMBBwNCAAQAwOGam7AKOi5LKmb9lK1rAzA2JTppqrFiIaUdjqmH
ZICJP00Wt0dfqOtEjgMEv1Hhu1DmKZn2ehvpxwPSzBr5o1UwUzBRBgNVHREBAf8E
RzBFgkNCNkw4YlI0UldJRU42NUZLamdUTzV1YmRvNUZWdkpNS2lxdjFZeTRULnNl
bGYtc2lnbmVkLmdvYXV0aGVudGlrLmlvMAoGCCqGSM49BAMCA0gAMEUCIC/JAfnl
uC30ihqepbiMCaTaPMbL8Ka2Lk92IYfMhf46AiEAz9Kmv6HF2D4MK54iwhz2WqvF
8vo+OiGdTQ1Qoj7fgYU=
-----END CERTIFICATE-----
"""
TEST_CORDS_KEY = """
-----BEGIN EC PRIVATE KEY-----
MHcCAQEEIKy6mPLJc5v71InMMvYaxyXI3xXpwQTPLyAYWVFnZHVioAoGCCqGSM49
AwEHoUQDQgAEAMDhmpuwCjouSypm/ZStawMwNiU6aaqxYiGlHY6ph2SAiT9NFrdH
X6jrRI4DBL9R4btQ5imZ9nob6ccD0swa+Q==
-----END EC PRIVATE KEY-----
"""
class TestJWKS(OAuthTestCase): class TestJWKS(OAuthTestCase):
"""Test JWKS view""" """Test JWKS view"""
@ -29,6 +57,8 @@ class TestJWKS(OAuthTestCase):
body = json.loads(response.content.decode()) body = json.loads(response.content.decode())
self.assertEqual(len(body["keys"]), 1) self.assertEqual(len(body["keys"]), 1)
PyJWKSet.from_dict(body) PyJWKSet.from_dict(body)
key = body["keys"][0]
load_der_x509_certificate(base64.b64decode(key["x5c"][0]), default_backend()).public_key()
def test_hs256(self): def test_hs256(self):
"""Test JWKS request with HS256""" """Test JWKS request with HS256"""
@ -60,3 +90,25 @@ class TestJWKS(OAuthTestCase):
body = json.loads(response.content.decode()) body = json.loads(response.content.decode())
self.assertEqual(len(body["keys"]), 1) self.assertEqual(len(body["keys"]), 1)
PyJWKSet.from_dict(body) PyJWKSet.from_dict(body)
def test_ecdsa_coords_mismatched(self):
"""Test JWKS request with ES256"""
cert = CertificateKeyPair.objects.create(
name=generate_id(),
key_data=TEST_CORDS_KEY,
certificate_data=TEST_CORDS_CERT,
)
provider = OAuth2Provider.objects.create(
name="test",
client_id="test",
authorization_flow=create_test_flow(),
redirect_uris="http://local.invalid",
signing_key=cert,
)
app = Application.objects.create(name="test", slug="test", provider=provider)
response = self.client.get(
reverse("authentik_providers_oauth2:jwks", kwargs={"application_slug": app.slug})
)
body = json.loads(response.content.decode())
self.assertEqual(len(body["keys"]), 1)
PyJWKSet.from_dict(body)

View File

@ -15,27 +15,49 @@ from cryptography.hazmat.primitives.serialization import Encoding
from django.http import HttpRequest, HttpResponse, JsonResponse from django.http import HttpRequest, HttpResponse, JsonResponse
from django.shortcuts import get_object_or_404 from django.shortcuts import get_object_or_404
from django.views import View from django.views import View
from jwt.utils import base64url_encode
from authentik.core.models import Application from authentik.core.models import Application
from authentik.crypto.models import CertificateKeyPair from authentik.crypto.models import CertificateKeyPair
from authentik.providers.oauth2.models import JWTAlgorithms, OAuth2Provider from authentik.providers.oauth2.models import JWTAlgorithms, OAuth2Provider
def b64_enc(number: int) -> str:
"""Convert number to base64-encoded octet-value"""
length = ((number).bit_length() + 7) // 8
number_bytes = number.to_bytes(length, "big")
final = urlsafe_b64encode(number_bytes).rstrip(b"=")
return final.decode("ascii")
# See https://notes.salrahman.com/generate-es256-es384-es512-private-keys/ # See https://notes.salrahman.com/generate-es256-es384-es512-private-keys/
# and _CURVE_TYPES in the same file as the below curve files # and _CURVE_TYPES in the same file as the below curve files
ec_crv_map = { ec_crv_map = {
SECP256R1: "P-256", SECP256R1: "P-256",
SECP384R1: "P-384", SECP384R1: "P-384",
SECP521R1: "P-512", SECP521R1: "P-521",
} }
min_length_map = {
SECP256R1: 32,
SECP384R1: 48,
SECP521R1: 66,
}
# https://github.com/jpadilla/pyjwt/issues/709
def bytes_from_int(val: int, min_length: int = 0) -> bytes:
"""Custom bytes_from_int that accepts a minimum length"""
remaining = val
byte_length = 0
while remaining != 0:
remaining >>= 8
byte_length += 1
length = max([byte_length, min_length])
return val.to_bytes(length, "big", signed=False)
def to_base64url_uint(val: int, min_length: int = 0) -> bytes:
"""Custom to_base64url_uint that accepts a minimum length"""
if val < 0:
raise ValueError("Must be a positive integer")
int_bytes = bytes_from_int(val, min_length)
if len(int_bytes) == 0:
int_bytes = b"\x00"
return base64url_encode(int_bytes)
class JWKSView(View): class JWKSView(View):
@ -55,34 +77,33 @@ class JWKSView(View):
"kty": "RSA", "kty": "RSA",
"alg": JWTAlgorithms.RS256, "alg": JWTAlgorithms.RS256,
"use": "sig", "use": "sig",
"n": b64_enc(public_numbers.n), "n": to_base64url_uint(public_numbers.n).decode(),
"e": b64_enc(public_numbers.e), "e": to_base64url_uint(public_numbers.e).decode(),
} }
elif isinstance(private_key, EllipticCurvePrivateKey): elif isinstance(private_key, EllipticCurvePrivateKey):
public_key: EllipticCurvePublicKey = private_key.public_key() public_key: EllipticCurvePublicKey = private_key.public_key()
public_numbers = public_key.public_numbers() public_numbers = public_key.public_numbers()
curve_type = type(public_key.curve)
key_data = { key_data = {
"kid": key.kid, "kid": key.kid,
"kty": "EC", "kty": "EC",
"alg": JWTAlgorithms.ES256, "alg": JWTAlgorithms.ES256,
"use": "sig", "use": "sig",
"x": b64_enc(public_numbers.x), "x": to_base64url_uint(public_numbers.x, min_length_map[curve_type]).decode(),
"y": b64_enc(public_numbers.y), "y": to_base64url_uint(public_numbers.y, min_length_map[curve_type]).decode(),
"crv": ec_crv_map.get(type(public_key.curve), public_key.curve.name), "crv": ec_crv_map.get(curve_type, public_key.curve.name),
} }
else: else:
return key_data return key_data
key_data["x5c"] = [b64encode(key.certificate.public_bytes(Encoding.DER)).decode("utf-8")] key_data["x5c"] = [b64encode(key.certificate.public_bytes(Encoding.DER)).decode("utf-8")]
key_data["x5t"] = ( key_data["x5t"] = urlsafe_b64encode(
urlsafe_b64encode(key.certificate.fingerprint(hashes.SHA1())) # nosec key.certificate.fingerprint(hashes.SHA1())
.decode("utf-8") ).decode( # nosec
.rstrip("=") "utf-8"
) ).rstrip("=")
key_data["x5t#S256"] = ( key_data["x5t#S256"] = urlsafe_b64encode(
urlsafe_b64encode(key.certificate.fingerprint(hashes.SHA256())) key.certificate.fingerprint(hashes.SHA256())
.decode("utf-8") ).decode("utf-8").rstrip("=")
.rstrip("=")
)
return key_data return key_data
def get(self, request: HttpRequest, application_slug: str) -> HttpResponse: def get(self, request: HttpRequest, application_slug: str) -> HttpResponse:

View File

@ -52,7 +52,7 @@ import TabItem from "@theme/TabItem";
OPENID_AUTHORIZATION_ENDPOINT: https://authentik.company/application/o/authorize/ OPENID_AUTHORIZATION_ENDPOINT: https://authentik.company/application/o/authorize/
OPENID_CLIENT_ID: # client ID from above OPENID_CLIENT_ID: # client ID from above
OPENID_ISSUER: https://authentik.company/application/o/*Slug of the application from above*/ OPENID_ISSUER: https://authentik.company/application/o/*Slug of the application from above*/
OPENID_JWKS_ENDPOINT: https://authentik.company/application/o/*Slug of the application from above*/jwks/ OPENID_JWKS_ENDPOINT: https://authentik.company/application/o/*Slug of the application from above*/jwks/?exclude_x5
OPENID_REDIRECT_URI: https://guacamole.company/ # This must match the redirect URI above OPENID_REDIRECT_URI: https://guacamole.company/ # This must match the redirect URI above
``` ```
@ -64,7 +64,7 @@ OPENID_REDIRECT_URI: https://guacamole.company/ # This must match the redirect U
openid-authorization-endpoint=https://authentik.company/application/o/authorize/ openid-authorization-endpoint=https://authentik.company/application/o/authorize/
openid-client-id=# client ID from above openid-client-id=# client ID from above
openid-issuer=https://authentik.company/application/o/*Slug of the application from above*/ openid-issuer=https://authentik.company/application/o/*Slug of the application from above*/
openid-jwks-endpoint=https://authentik.company/application/o/*Slug of the application from above*/jwks/ openid-jwks-endpoint=https://authentik.company/application/o/*Slug of the application from above*/jwks/?exclude_x5
openid-redirect-uri=https://guacamole.company/ # This must match the redirect URI above openid-redirect-uri=https://guacamole.company/ # This must match the redirect URI above
``` ```