From edd4f4caf3cde449c73ffd535da72b45e5c1435a Mon Sep 17 00:00:00 2001
From: Julian Rother <julian@cccv.de>
Date: Fri, 22 Mar 2024 00:09:07 +0100
Subject: [PATCH] OpenID Connect Core 1.0 and Discovery 1.0 support
Limited to OpenID provider conformance profiles "Basic" and "Config":
- Support for features mandatory to implement for all OpenID Providers,
not the feature set for Dynamic OpenID Providers
- Only Authorization Code Flow, no support for Implicit/Hybrid Flow
- Only code response type, no support for token/id_token
- Server metadata is served at /.well-known/openid-configuration
Additional/optional features:
- Support for "claims" parameter
- Support for standard scopes "profile" and "email"
- Support for non-standard scope/claim "groups" (in violation of RFC 9068)
Compatability with existing (working) uffd client setups: Authorization
requests without the "openid" scope behave the same as before Prior to this
change authorization requests with the "openid" scope were rejected by uffd.
This change adds direct dependencies to pyjwt and cryptography. Prior to this
change both were already transitive dependencies of oauthlib.
---
README.md | 47 +-
debian/control | 3 +-
setup.py | 5 +-
tests/migrations/test_fuzzy.py | 4 +-
tests/models/test_oauth2.py | 158 ++++
tests/views/test_oauth2.py | 705 +++++++++++++++++-
.../01fdd7820f29_openid_connect_support.py | 148 ++++
uffd/models/__init__.py | 2 +-
uffd/models/oauth2.py | 203 ++++-
uffd/models/service.py | 10 +
uffd/views/oauth2.py | 661 ++++++++++------
11 files changed, 1701 insertions(+), 245 deletions(-)
create mode 100644 tests/models/test_oauth2.py
create mode 100644 uffd/migrations/versions/01fdd7820f29_openid_connect_support.py
diff --git a/README.md b/README.md
index f1cd4f30..2c51d402 100644
--- a/README.md
+++ b/README.md
@@ -15,7 +15,8 @@ Please note that we refer to Debian packages here and **not** pip packages.
- python3-qrcode
- python3-fido2 (version 0.5.0 or 0.9.1, optional)
- python3-prometheus-client (optional, needed for metrics)
-- python3-oauthlib
+- python3-jwt
+- python3-cryptography
- python3-flask-babel
- python3-argon2
- python3-itsdangerous (also a dependency of python3-flask)
@@ -97,6 +98,8 @@ The services need to be setup to use the following URLs with the Authorization C
* `/oauth2/token`: token request endpoint
* `/oauth2/userinfo`: endpoint that provides information about the current user
+If the service supports server metadata discovery ([RFC 8414](https://www.rfc-editor.org/rfc/rfc8414)), configuring the base url of your uffd installation or `/.well-known/openid-configuration` as the discovery endpoint should be sufficient.
+
The only OAuth2 scope supported is `profile`. The userinfo endpoint returns json data with the following structure:
```
@@ -114,6 +117,48 @@ The only OAuth2 scope supported is `profile`. The userinfo endpoint returns json
`id` is the numeric (Unix) user id, `name` the display name and `nickname` the loginname of the user.
+## OpenID Connect Single-Sign-On Provider
+
+In addition to plain OAuth2, uffd also has basic OpenID Connect support.
+Endpoint URLs are the same as for plain OAuth2.
+OpenID Connect support is enabled by requesting the `openid` scope.
+ID token signing keys are served at `/oauth2/keys`.
+
+See [OpenID Connect Core 1.0](https://openid.net/specs/openid-connect-core-1_0.html) specification for more details.
+
+Supported flows and response types:
+
+* Only Authorization Code Flow with `code` response type
+
+Supported scopes:
+
+* `openid`: Enables OpenID Connect support and returns mandatory `sub` claim
+* `profile`: Returns `name` and `preferred_username` claims
+* `email`: Returns `email` and `email_verified` claims
+* `groups`: Returns non-standard `groups` claim
+
+Supported claims:
+
+* `sub` (string): Decimal encoded numeric (Unix) user id
+* `name` (string): Display name
+* `preferred_username`(string): Loginname
+* `email` (string): Service-specific or primary email address
+* `email_verified` (boolean): Verification status of `email` value (always `true`)
+* `groups` (array of strings): Names of groups the user is a member of (non-standard)
+
+uffd supports the optional `claims` authorization request parameter for requesting claims individually.
+
+Note that there is a IANA-registered `groups` claim with a syntax borrowed from [SCIM](https://www.rfc-editor.org/rfc/rfc7643.html).
+The syntax used by uffd is different and incompatible, although arguably more common for a claim named "groups" in this context.
+
+uffd aims for complience with OpenID provider conformance profiles Basic and Config.
+It is, however, not a certified OpenID provider and it has the following limitations:
+
+* Only the `none` value for the `prompt` authorization request parameter is recognized. Other values (`login`, `consent` and `select_account`) are ignored.
+* The `max_age` authorization request parameter is not supported and ignored by uffd.
+* The `auth_time` claim is not supported and neither returned if the `max_age` authorization request parameter is present nor if it is requested via the `claims` parameter.
+* Requesting the `sub` claim with a specific value for the ID Token (or passing the `id_token_hint` authorization request parameter) is only supported if the `prompt` authorization request parameter is set to `none`. The authorization request is rejected otherwise.
+
## Metrics
Uffd can export metrics in a prometheus compatible way. It needs python3-prometheus-client for this feature to work.
diff --git a/debian/control b/debian/control
index 0e498d6a..a9a87dd5 100644
--- a/debian/control
+++ b/debian/control
@@ -23,7 +23,8 @@ Depends:
python3-flask-migrate,
python3-qrcode,
python3-fido2,
- python3-oauthlib,
+ python3-jwt,
+ python3-cryptography,
python3-flask-babel,
python3-argon2,
python3-itsdangerous,
diff --git a/setup.py b/setup.py
index 794373b3..189b3010 100644
--- a/setup.py
+++ b/setup.py
@@ -36,7 +36,8 @@ setup(
'Flask-SQLAlchemy==2.1',
'qrcode==6.1',
'fido2==0.5.0',
- 'oauthlib==2.1.0',
+ 'cryptography==2.6.1',
+ 'pyjwt==1.7.0',
'Flask-Migrate==2.1.1',
'Flask-Babel==0.11.2',
'alembic==1.0.0',
@@ -52,11 +53,9 @@ setup(
'cffi # v1.12.2 no longer works with python3.9. Newer versions seem to work fine.',
'chardet==3.0.4',
'click==7.0',
- 'cryptography==2.6.1',
'idna==2.6',
'Jinja2==2.10',
'MarkupSafe==1.1.0',
- 'oauthlib==2.1.0',
'pyasn1==0.4.2',
'pycparser==2.19',
'requests==2.21.0',
diff --git a/tests/migrations/test_fuzzy.py b/tests/migrations/test_fuzzy.py
index 7200267b..226e75ee 100644
--- a/tests/migrations/test_fuzzy.py
+++ b/tests/migrations/test_fuzzy.py
@@ -60,8 +60,8 @@ class TestFuzzy(MigrationTestCase):
service = Service(name='testservice', access_group=group)
oauth2_client = OAuth2Client(service=service, client_id='testclient', client_secret='testsecret', redirect_uris=['http://localhost:1234/callback'], logout_uris=[OAuth2LogoutURI(method='GET', uri='http://localhost:1234/callback')])
db.session.add_all([service, oauth2_client])
- db.session.add(OAuth2Grant(user=user, client=oauth2_client, code='testcode', redirect_uri='http://example.com/callback', expires=datetime.datetime.now()))
- db.session.add(OAuth2Token(user=user, client=oauth2_client, token_type='Bearer', access_token='testcode', refresh_token='testcode', expires=datetime.datetime.now()))
+ db.session.add(OAuth2Grant(user=user, client=oauth2_client, _code='testcode', redirect_uri='http://example.com/callback', expires=datetime.datetime.now()))
+ db.session.add(OAuth2Token(user=user, client=oauth2_client, token_type='Bearer', _access_token='testcode', _refresh_token='testcode', expires=datetime.datetime.now()))
db.session.add(OAuth2DeviceLoginInitiation(client=oauth2_client, confirmations=[DeviceLoginConfirmation(user=user)]))
db.session.add(PasswordToken(user=user))
db.session.commit()
diff --git a/tests/models/test_oauth2.py b/tests/models/test_oauth2.py
new file mode 100644
index 00000000..52b66299
--- /dev/null
+++ b/tests/models/test_oauth2.py
@@ -0,0 +1,158 @@
+import unittest
+import datetime
+
+import jwt
+
+from uffd.database import db
+from uffd.models import OAuth2Key
+
+from tests.utils import UffdTestCase
+
+TEST_JWK = dict(
+ id='HvOn74G7njK1GoFNe8Dta087casdWMsm06pNhOXRgJU',
+ created=datetime.datetime(2023, 11, 9, 0, 21, 10),
+ active=True,
+ algorithm='RS256',
+ private_key_jwk='''{
+ "kty": "RSA",
+ "key_ops": ["sign"],
+ "n": "vrznqUy8Xamph6s0Z02fFMIyjwLAMio35i9DXYjXP1ZQwSZ3SsIh3m2ablMnlu8PVlnYUzoj8rXyAWND0FSfWoQQxv1rq15pllKueddLoJsv321N_NRB8beGsLrsndw8QO0q3RWqV9O3kqhlTMjgj6bquX42wLaXrPLJyfbT3zObBsToG4UxpOyly84aklJXU5wIs0cbmjbfd8Xld38BG8Oh7Ozy5b93vPpJW6rudZRxU6QYC0r9bFFLIHJWrR4bzQMLGoJ63xjPOCl4WNpOYc9B7PNgnWTLXlFd51Hw9CaT2MRWsKNCSU77f6nZkfjWa1IsQdF0I48m46qgq7bEOOl9DbThbCnpblWrctdyg6du-OvCyVmkAo1KGtANl0027pgqUI_9HBMi33y3UPQm1ALHXIyIDBZtExH3lD6MMK3XGJfUxZuIOBndK-PXm5Fed52bgLOcf-24X6aHFn-8oyDVIj9OHkKWjy7jtKdmqZc4pBdVuCaMCYzj8iERWA3H",
+ "e": "AQAB",
+ "d": "G7yoH5mLcZTA6ia-byCoN-zpofGvdga9AZnxPO0vsq6K_cY_O2gxuVZ3n6reAKKbuLNGCbb_D_Dffs4q8rprlfkgi3TCLzXX5Zv5HWTD7a4Y7xpxEzQ2sWo-iagVIqZVPh0pyjliqnTyUWnFmWiY0gBe9UHianHjFVZqe8E2HFOKgW3UUbQz0keg8JtJ3T9gzZrM38KWbqhOJO0VVSRAoANPTSnumfRsUCyWywrMtIfgAbQaKazqX3xkOsAF1L-iNfd6slzPvRyIQVflVDMdfKnsu-lHiKJ0DK_lg9f55T5FymgcXsq43EKBQ2H4v2dafIm-vtWx_TRZWj_msD32BEPBA-zTqh_oP1r6a3DZh4DBtWY3vzSiuhAC0erlRs-hRTX_e9ET5fUbJnmNxjnxQD9zZmwq4ujMK6KFnHct8t77Qxj3a-wDR_XyDJ4_EKYqHlcVHfxGNBSvIdjuZJkPJnVpVtfCtpyamQIR4u5oNV7fIwYe_tFnw0Y90rGoJMzB",
+ "p": "-A-FnH21HJ7GPWUm9k3mxsxSchy89QEUCZZiH6EcB4ZP8wJsxrQsUSIHCR74YmZEI3Ulsum1Ql4x50k7Q2sNh9SnwKvmctjksehGy4yCrdunAqjqyz3wFwGaKWnhn3frkiqH5ATjkOoc8qHz8saa7reeVClj47ZWyy-Nl559ycLMs0rI1N_THzO07C3jSbJhyPj0yeygAflsRqqnNvEQ6ps1VLiqf9G5jfSvUUn5DyKIpep9iGo29caGSIPIy_2h",
+ "q": "xNe1-QWskxOcY_GiHpFWdvzqr1o9fxg5whgpNcGi3caokw2iNHRYut4cbVvFFBlv_9B5QCl9WVfR2ADG0AtvkvUxEZqCdxEvcqjIANeRLKHDjW5kMuPS0_fcskFP-r7mCM9SBfPplfMVCF5nuNWf5LzNopWfsTChIDD1rSpPjItNYuwLXszm_3R81HHHeQLcyvoMxLCmeLy5TXX2hXOMHh2IMZCXAHopJmLJUVnQ48kr5jd2l0kLbmx3aBqdccJn",
+ "dp": "MLS7g1KbcRcrzXpDADGjkn0j4wwJfgHMMWW5toQnwMJ6iDh9qzZNTVDlGMFf-9IgpuWllU-WK4XbPpJ-dGpcqcLzfT1DbmFv5g65d9YLAqASVs9b6rQqpBnIb0E-79TYCEcZj4f2NsoBDRMHly-v1BdxmwzVdCylNhgMMS0Jfcgl8T5J2KJqDcJVT9piumGwGYnoZo1zjW-v9uAjHQKQU8BN5Git8ZL4YAsfMVLY-EPLmOhF5bcVO4TTcQGPN56B",
+ "dq": "HiiSl-G3rB0QE_v8g8Ruw_JCHrWrwGI8zzEWd0cApgv-3fDzzieZRKAtKNArpMW09DPDsAHrU5nx669KxqtJ3_EzIGhU3ttCMsYLRp3Af18VcADe1zEypwlNxf3dvCQtaGIjRgg13KSOr2aPa7FHOyt2MhfMjMBPn3gA3BQkdfsN0z8pCtBIABGf4ojAMBkxLOQcurH5_3uixGxzZcTrTd3mdPmbORZ-YYQ3JgCl0ZCL6kzLHaiyWKvDq66QOtK3",
+ "qi": "ySqD9cUxbq3wkCsPQId_YfQLIqb5RK_JJIMjtBOdTdo4aT5tmodYCSmjBmhrYXjDWtyJdelvPfdSfgncHJhf8VgkZ8TPvUeaQwsQFBwB5llwpdb72eEEJrmG1SVwNMoFCLXdNT3ACad16cUDMnWmklH0X07OzdxGOBnGhgLZUs4RbPjLH7OpYTyQqVy2L8vofqJR42cfePZw8WQM4k0PPbhralhybExIkSCmaQyYbACZ5k0OVQErEqnj4elglA0h"
+ }''',
+ public_key_jwk='''{
+ "kty": "RSA",
+ "key_ops": ["verify"],
+ "n": "vrznqUy8Xamph6s0Z02fFMIyjwLAMio35i9DXYjXP1ZQwSZ3SsIh3m2ablMnlu8PVlnYUzoj8rXyAWND0FSfWoQQxv1rq15pllKueddLoJsv321N_NRB8beGsLrsndw8QO0q3RWqV9O3kqhlTMjgj6bquX42wLaXrPLJyfbT3zObBsToG4UxpOyly84aklJXU5wIs0cbmjbfd8Xld38BG8Oh7Ozy5b93vPpJW6rudZRxU6QYC0r9bFFLIHJWrR4bzQMLGoJ63xjPOCl4WNpOYc9B7PNgnWTLXlFd51Hw9CaT2MRWsKNCSU77f6nZkfjWa1IsQdF0I48m46qgq7bEOOl9DbThbCnpblWrctdyg6du-OvCyVmkAo1KGtANl0027pgqUI_9HBMi33y3UPQm1ALHXIyIDBZtExH3lD6MMK3XGJfUxZuIOBndK-PXm5Fed52bgLOcf-24X6aHFn-8oyDVIj9OHkKWjy7jtKdmqZc4pBdVuCaMCYzj8iERWA3H",
+ "e": "AQAB"
+ }''',
+)
+
+class TestOAuth2Key(UffdTestCase):
+ def setUp(self):
+ super().setUp()
+ db.session.add(OAuth2Key(**TEST_JWK))
+ db.session.add(OAuth2Key(
+ id='1e9gdk7',
+ created=datetime.datetime(2014, 11, 8, 0, 0, 0),
+ active=True,
+ algorithm='RS256',
+ private_key_jwk='invalid',
+ public_key_jwk='''{
+ "kty":"RSA",
+ "n":"w7Zdfmece8iaB0kiTY8pCtiBtzbptJmP28nSWwtdjRu0f2GFpajvWE4VhfJAjEsOcwYzay7XGN0b-X84BfC8hmCTOj2b2eHT7NsZegFPKRUQzJ9wW8ipn_aDJWMGDuB1XyqT1E7DYqjUCEOD1b4FLpy_xPn6oV_TYOfQ9fZdbE5HGxJUzekuGcOKqOQ8M7wfYHhHHLxGpQVgL0apWuP2gDDOdTtpuld4D2LK1MZK99s9gaSjRHE8JDb1Z4IGhEcEyzkxswVdPndUWzfvWBBWXWxtSUvQGBRkuy1BHOa4sP6FKjWEeeF7gm7UMs2Nm2QUgNZw6xvEDGaLk4KASdIxRQ",
+ "e":"AQAB"
+ }'''
+ ))
+ db.session.commit()
+ self.key = OAuth2Key.query.get('HvOn74G7njK1GoFNe8Dta087casdWMsm06pNhOXRgJU')
+ self.key_oidc_spec = OAuth2Key.query.get('1e9gdk7')
+
+ def test_private_key(self):
+ self.key.private_key
+
+ def test_public_key(self):
+ self.key.private_key
+
+ def test_public_key_jwks_dict(self):
+ self.assertEqual(self.key.public_key_jwks_dict, {
+ "kid": "HvOn74G7njK1GoFNe8Dta087casdWMsm06pNhOXRgJU",
+ "kty": "RSA",
+ "alg": "RS256",
+ "use": "sig",
+ "n": "vrznqUy8Xamph6s0Z02fFMIyjwLAMio35i9DXYjXP1ZQwSZ3SsIh3m2ablMnlu8PVlnYUzoj8rXyAWND0FSfWoQQxv1rq15pllKueddLoJsv321N_NRB8beGsLrsndw8QO0q3RWqV9O3kqhlTMjgj6bquX42wLaXrPLJyfbT3zObBsToG4UxpOyly84aklJXU5wIs0cbmjbfd8Xld38BG8Oh7Ozy5b93vPpJW6rudZRxU6QYC0r9bFFLIHJWrR4bzQMLGoJ63xjPOCl4WNpOYc9B7PNgnWTLXlFd51Hw9CaT2MRWsKNCSU77f6nZkfjWa1IsQdF0I48m46qgq7bEOOl9DbThbCnpblWrctdyg6du-OvCyVmkAo1KGtANl0027pgqUI_9HBMi33y3UPQm1ALHXIyIDBZtExH3lD6MMK3XGJfUxZuIOBndK-PXm5Fed52bgLOcf-24X6aHFn-8oyDVIj9OHkKWjy7jtKdmqZc4pBdVuCaMCYzj8iERWA3H",
+ "e": "AQAB"
+ })
+
+ def test_encode_jwt(self):
+ jwtdata = self.key.encode_jwt({'aud': 'test', 'foo': 'bar'})
+ self.assertEqual(
+ jwt.get_unverified_header(jwtdata),
+ # typ is optional, x5u/x5c/jku/jwk are discoraged by OIDC Core 1.0 spec section 2
+ {'kid': self.key.id, 'alg': self.key.algorithm, 'typ': 'JWT'}
+ )
+ self.assertEqual(
+ OAuth2Key.decode_jwt(jwtdata, audience='test'),
+ {'aud': 'test', 'foo': 'bar'}
+ )
+ self.key.active = False
+ with self.assertRaises(jwt.exceptions.InvalidKeyError):
+ self.key.encode_jwt({'aud': 'test', 'foo': 'bar'})
+
+ def test_oidc_hash(self):
+ # Example from OIDC Core 1.0 spec A.3
+ self.assertEqual(
+ self.key.oidc_hash(b'jHkWEdUXMU1BwAsC4vtUsZwnNvTIxEl0z9K3vx5KF0Y'),
+ '77QmUPtjPfzWtF2AnpK9RQ'
+ )
+ # Example from OIDC Core 1.0 spec A.4
+ self.assertEqual(
+ self.key.oidc_hash(b'Qcb0Orv1zh30vL1MPRsbm-diHiMwcLyZvn1arpZv-Jxf_11jnpEX3Tgfvk'),
+ 'LDktKdoQak3Pk0cnXxCltA'
+ )
+ # Example from OIDC Core 1.0 spec A.6
+ self.assertEqual(
+ self.key.oidc_hash(b'jHkWEdUXMU1BwAsC4vtUsZwnNvTIxEl0z9K3vx5KF0Y'),
+ '77QmUPtjPfzWtF2AnpK9RQ'
+ )
+ self.assertEqual(
+ self.key.oidc_hash(b'Qcb0Orv1zh30vL1MPRsbm-diHiMwcLyZvn1arpZv-Jxf_11jnpEX3Tgfvk'),
+ 'LDktKdoQak3Pk0cnXxCltA'
+ )
+
+ def test_decode_jwt(self):
+ # Example from OIDC Core 1.0 spec A.2
+ jwt_data = (
+ 'eyJraWQiOiIxZTlnZGs3IiwiYWxnIjoiUlMyNTYifQ.ewogImlz'
+ 'cyI6ICJodHRwOi8vc2VydmVyLmV4YW1wbGUuY29tIiwKICJzdWIiOiAiMjQ4'
+ 'Mjg5NzYxMDAxIiwKICJhdWQiOiAiczZCaGRSa3F0MyIsCiAibm9uY2UiOiAi'
+ 'bi0wUzZfV3pBMk1qIiwKICJleHAiOiAxMzExMjgxOTcwLAogImlhdCI6IDEz'
+ 'MTEyODA5NzAsCiAibmFtZSI6ICJKYW5lIERvZSIsCiAiZ2l2ZW5fbmFtZSI6'
+ 'ICJKYW5lIiwKICJmYW1pbHlfbmFtZSI6ICJEb2UiLAogImdlbmRlciI6ICJm'
+ 'ZW1hbGUiLAogImJpcnRoZGF0ZSI6ICIwMDAwLTEwLTMxIiwKICJlbWFpbCI6'
+ 'ICJqYW5lZG9lQGV4YW1wbGUuY29tIiwKICJwaWN0dXJlIjogImh0dHA6Ly9l'
+ 'eGFtcGxlLmNvbS9qYW5lZG9lL21lLmpwZyIKfQ.rHQjEmBqn9Jre0OLykYNn'
+ 'spA10Qql2rvx4FsD00jwlB0Sym4NzpgvPKsDjn_wMkHxcp6CilPcoKrWHcip'
+ 'R2iAjzLvDNAReF97zoJqq880ZD1bwY82JDauCXELVR9O6_B0w3K-E7yM2mac'
+ 'AAgNCUwtik6SjoSUZRcf-O5lygIyLENx882p6MtmwaL1hd6qn5RZOQ0TLrOY'
+ 'u0532g9Exxcm-ChymrB4xLykpDj3lUivJt63eEGGN6DH5K6o33TcxkIjNrCD'
+ '4XB1CKKumZvCedgHHF3IAK4dVEDSUoGlH9z4pP_eWYNXvqQOjGs-rDaQzUHl'
+ '6cQQWNiDpWOl_lxXjQEvQ'
+ )
+ self.assertEqual(
+ OAuth2Key.decode_jwt(jwt_data, options={'verify_exp': False, 'verify_aud': False}),
+ {
+ "iss": "http://server.example.com",
+ "sub": "248289761001",
+ "aud": "s6BhdRkqt3",
+ "nonce": "n-0S6_WzA2Mj",
+ "exp": 1311281970,
+ "iat": 1311280970,
+ "name": "Jane Doe",
+ "given_name": "Jane",
+ "family_name": "Doe",
+ "gender": "female",
+ "birthdate": "0000-10-31",
+ "email": "janedoe@example.com",
+ "picture": "http://example.com/janedoe/me.jpg"
+ }
+ )
+ with self.assertRaises(jwt.exceptions.InvalidKeyError):
+ # {"alg":"RS256"} -> no key id
+ OAuth2Key.decode_jwt('eyJhbGciOiJSUzI1NiJ9.' + jwt_data.split('.', 1)[-1])
+ with self.assertRaises(jwt.exceptions.InvalidKeyError):
+ # {"kid":"XXXXX","alg":"RS256"} -> unknown key id
+ OAuth2Key.decode_jwt('eyJraWQiOiJYWFhYWCIsImFsZyI6IlJTMjU2In0.' + jwt_data.split('.', 1)[-1])
+ OAuth2Key.query.get('1e9gdk7').active = False
+ with self.assertRaises(jwt.exceptions.InvalidKeyError):
+ # not active
+ OAuth2Key.decode_jwt(jwt_data)
+
+ def test_generate_rsa_key(self):
+ key = OAuth2Key.generate_rsa_key()
+ self.assertEqual(key.algorithm, 'RS256')
diff --git a/tests/views/test_oauth2.py b/tests/views/test_oauth2.py
index 42067d58..2451af51 100644
--- a/tests/views/test_oauth2.py
+++ b/tests/views/test_oauth2.py
@@ -1,13 +1,16 @@
+import unittest
from urllib.parse import urlparse, parse_qs
+import jwt
from flask import url_for, session
from uffd.database import db
from uffd.password_hash import PlaintextPasswordHash
from uffd.remailer import remailer
-from uffd.models import DeviceLoginConfirmation, Service, OAuth2Client, OAuth2DeviceLoginInitiation, RemailerMode
+from uffd.models import DeviceLoginConfirmation, Service, OAuth2Client, OAuth2DeviceLoginInitiation, RemailerMode, OAuth2Key
from tests.utils import dump, UffdTestCase
+from tests.models.test_oauth2 import TEST_JWK
class TestViews(UffdTestCase):
def setUpDB(self):
@@ -69,6 +72,9 @@ class TestViews(UffdTestCase):
self.assertTrue(oauth2_client.client_secret.verify('testsecret'))
def test_authorization_without_redirect_uri(self):
+ client = OAuth2Client.query.filter_by(client_id='test').one()
+ client.redirect_uris.remove('http://localhost:5009/callback2')
+ db.session.commit()
self.login_as('user')
r = self.client.get(path=url_for('oauth2.authorize', response_type='code', client_id='test', state='teststate', scope='profile'), follow_redirects=False)
self.assert_authorization(r)
@@ -84,6 +90,15 @@ class TestViews(UffdTestCase):
r = self.client.get(path=url_for('oauth2.authorize', response_type='code', client_id='test', state='teststate', scope='', redirect_uri='http://localhost:5009/callback'), follow_redirects=False)
self.assert_authorization(r)
+ def test_authorization_access_denied(self):
+ client = OAuth2Client.query.filter_by(client_id='test').one()
+ client.service.limit_access = True
+ db.session.commit()
+ self.login_as('user')
+ r = self.client.get(path=url_for('oauth2.authorize', response_type='code', client_id='test', state='teststate', redirect_uri='http://localhost:5009/callback', scope='profile'), follow_redirects=False)
+ self.assertEqual(r.status_code, 403)
+ dump('oauth2_authorization_access_denied', r)
+
def test_authorization_invalid_scope(self):
self.login_as('user')
r = self.client.get(path=url_for('oauth2.authorize', response_type='code', client_id='test', state='teststate', redirect_uri='http://localhost:5009/callback', scope='invalid'), follow_redirects=False)
@@ -190,7 +205,7 @@ class TestViews(UffdTestCase):
def test_token_invalid_code(self):
r = self.client.post(path=url_for('oauth2.token'),
data={'grant_type': 'authorization_code', 'code': 'abcdef', 'redirect_uri': 'http://localhost:5009/callback', 'client_id': 'test', 'client_secret': 'testsecret'}, follow_redirects=True)
- self.assertIn(r.status_code, [400, 401]) # oauthlib behaviour changed between v2.1.0 and v3.1.0
+ self.assertEqual(r.status_code, 400)
self.assertEqual(r.content_type, 'application/json')
self.assertEqual(r.json['error'], 'invalid_grant')
@@ -203,7 +218,7 @@ class TestViews(UffdTestCase):
r = self.client.post(path=url_for('oauth2.token'),
data={'grant_type': 'authorization_code', 'code': code, 'redirect_uri': 'http://localhost:5009/callback'},
headers={'Authorization': f'Basic dGVzdDp0ZXN0c2VjcmV0'}, follow_redirects=True)
- self.assertIn(r.status_code, [400, 401]) # oauthlib behaviour changed between v2.1.0 and v3.1.0
+ self.assertEqual(r.status_code, 400)
self.assertEqual(r.json['error'], 'invalid_grant')
def test_token_invalid_client(self):
@@ -211,12 +226,14 @@ class TestViews(UffdTestCase):
data={'grant_type': 'authorization_code', 'code': self.get_auth_code(), 'redirect_uri': 'http://localhost:5009/callback', 'client_id': 'invalid_client', 'client_secret': 'invalid_client_secret'}, follow_redirects=True)
self.assertEqual(r.status_code, 401)
self.assertEqual(r.content_type, 'application/json')
+ self.assertEqual(r.json['error'], 'invalid_client')
def test_token_unauthorized_client(self):
r = self.client.post(path=url_for('oauth2.token'),
data={'grant_type': 'authorization_code', 'code': self.get_auth_code(), 'redirect_uri': 'http://localhost:5009/callback', 'client_id': 'test'}, follow_redirects=True)
self.assertEqual(r.status_code, 401)
self.assertEqual(r.content_type, 'application/json')
+ self.assertEqual(r.json['error'], 'invalid_client')
def test_token_unsupported_grant_type(self):
r = self.client.post(path=url_for('oauth2.token'),
@@ -231,15 +248,16 @@ class TestViews(UffdTestCase):
db.session.commit()
r = self.client.post(path=url_for('oauth2.token'),
data={'grant_type': 'authorization_code', 'code': code, 'redirect_uri': 'http://localhost:5009/callback', 'client_id': 'test', 'client_secret': 'testsecret'}, follow_redirects=True)
- self.assertIn(r.status_code, [400, 401]) # oauthlib behaviour changed between v2.1.0 and v3.1.0
+ self.assertEqual(r.status_code, 400)
self.assertEqual(r.content_type, 'application/json')
+ self.assertEqual(r.json['error'], 'invalid_grant')
def test_userinfo_invalid_access_token(self):
token = 'invalidtoken'
r = self.client.get(path=url_for('oauth2.userinfo'), headers=[('Authorization', 'Bearer %s'%token)], follow_redirects=True)
self.assertEqual(r.status_code, 401)
- def test_userinfo_invalid_access_token(self):
+ def test_userinfo_deactivated_user(self):
r = self.client.post(path=url_for('oauth2.token'),
data={'grant_type': 'authorization_code', 'code': self.get_auth_code(), 'redirect_uri': 'http://localhost:5009/callback', 'client_id': 'test', 'client_secret': 'testsecret'}, follow_redirects=True)
token = r.json['access_token']
@@ -247,3 +265,680 @@ class TestViews(UffdTestCase):
db.session.commit()
r = self.client.get(path=url_for('oauth2.userinfo'), headers=[('Authorization', 'Bearer %s'%token)], follow_redirects=True)
self.assertEqual(r.status_code, 401)
+
+class TestOIDCConfigurationProfile(UffdTestCase):
+ def setUpDB(self):
+ db.session.add(OAuth2Key(**TEST_JWK))
+
+ def test_discover_spec(self):
+ ISSUER = 'https://sso.example.com'
+ r = self.client.get(base_url=ISSUER, path='/.well-known/openid-configuration')
+
+ # OIDC Discovery 1.0 section 4.2:
+ # > A successful response MUST use the 200 OK HTTP status code and return a
+ # > JSON object using the application/json content type that contains a set
+ # > of Claims as its members that are a subset of the Metadata values defined
+ # > in Section 3.
+ self.assertEqual(r.status_code, 200)
+ self.assertEqual(r.content_type, 'application/json')
+ self.assertIsInstance(r.json, dict) # also validates JSON syntax
+
+ # OIDC Discovery 1.0 section 4.2:
+ # > Claims that return multiple values are represented as JSON arrays.
+ # > Claims with zero elements MUST be omitted from the response.
+ for key, value in r.json.items():
+ if isinstance(value, list):
+ self.assertNotEqual(len(value), 0)
+
+ # OIDC Discovery 1.0 section 3 (REQUIRED metadata values)
+ required_fields = {'issuer', 'authorization_endpoint', 'jwks_uri', 'response_types_supported', 'subject_types_supported', 'id_token_signing_alg_values_supported'}
+ if 'code' in r.json.get('response_types_supported', []):
+ required_fields.add('token_endpoint')
+ if 'authorization_code' in r.json.get('grant_types_supported', ['authorization_code', 'implicit']):
+ required_fields.add('token_endpoint')
+ for field in required_fields:
+ self.assertIn(field, r.json)
+
+ # OIDC Discovery 1.0 section 3 (metadata value types)
+ bool_fields = ('claims_parameter_supported', 'request_parameter_supported', 'request_uri_parameter_supported', 'require_request_uri_registration')
+ list_fields = ('scopes_supported', 'response_types_supported', 'response_modes_supported', 'grant_types_supported', 'acr_values_supported', 'subject_types_supported', 'id_token_signing_alg_values_supported', 'id_token_encryption_alg_values_supported', 'id_token_encryption_enc_values_supported', 'userinfo_signing_alg_values_supported', 'userinfo_encryption_alg_values_supported', 'userinfo_encryption_enc_values_supported', 'request_object_signing_alg_values_supported', 'request_object_encryption_alg_values_supported', 'request_object_encryption_enc_values_supported', 'token_endpoint_auth_methods_supported', 'token_endpoint_auth_signing_alg_values_supported', 'display_values_supported', 'claim_types_supported', 'claims_supported', 'claims_locales_supported', 'ui_locales_supported')
+ https_url_fields = ('issuer', 'authorization_endpoint', 'token_endpoint', 'userinfo_endpoint', 'jwks_uri', 'registration_endpoint')
+ url_fields = ('service_documentation', 'op_policy_uri', 'op_tos_uri')
+ for field in bool_fields:
+ if field in r.json:
+ self.assertIsInstance(r.json[field], bool)
+ for field in list_fields:
+ if field in r.json:
+ self.assertIsInstance(r.json[field], list)
+ for field in https_url_fields:
+ if field in r.json:
+ self.assertIsInstance(r.json[field], str)
+ self.assertTrue(r.json[field].lower().startswith('https://'))
+ for field in url_fields:
+ if field in r.json:
+ self.assertIsInstance(r.json[field], str)
+ self.assertTrue(r.json[field].lower().startswith('http'))
+
+ # OIDC Discovery 1.0 section 3 (MUSTs on metadata values except https scheme and jwks_uri)
+ self.assertEqual(r.json['issuer'], ISSUER)
+ if 'scopes_supported' in r.json:
+ self.assertIsInstance(r.json['scopes_supported'], list)
+ for item in r.json['scopes_supported']:
+ self.assertIsInstance(item, str)
+ self.assertRegex(item, r'^[!#-\[\]-~]+$') # 1*( %x21 / %x23-5B / %x5D-7E )
+ self.assertIn('openid', r.json['scopes_supported'])
+ self.assertIn('RS256', r.json['id_token_signing_alg_values_supported'])
+ if 'token_endpoint_auth_signing_alg_values_supported' in r.json:
+ self.assertNotIn('none', r.json['token_endpoint_auth_signing_alg_values_supported'])
+
+ # OIDC Discovery 1.0 section 3 (jwks_uri) and RFC7517
+ self.assertTrue(r.json['jwks_uri'].startswith(ISSUER)) # Not a requirement by spec, but technically neccessary for this test
+ r_jwks = self.client.get(base_url=ISSUER, path=r.json['jwks_uri'][len(ISSUER):])
+ self.assertEqual(r_jwks.status_code, 200)
+ # The jwks_uri SHOULD include a Cache-Control header in the response that contains a max-age directive ...
+ self.assertIn('Cache-Control', r_jwks.headers)
+ self.assertIsInstance(r_jwks.json, dict) # also validates JSON syntax
+ self.assertIn('keys', r_jwks.json)
+ self.assertIsInstance(r_jwks.json['keys'], list)
+ has_sign_keys = False
+ has_encrypt_keys = False
+ kids = set()
+ for key in r_jwks.json['keys']:
+ self.assertIn('kty', key)
+ self.assertIsInstance(key['kty'], str)
+ if 'use' in key:
+ self.assertIsInstance(key['use'], str)
+ if key['use'] == 'sig':
+ has_sign_keys = True
+ if key['use'] == 'enc':
+ has_enc_keys = True
+ if 'key_ops' in key:
+ self.assertIsInstance(key['key_ops'], list)
+ self.assertNotIn('use', key) # SHOULD
+ for value in key['key_ops']:
+ self.assertIsInstance(value, str)
+ self.assertEqual(key['key_ops'].count(value), 1)
+ # OIDC: "The JWK Set MUST NOT contain private or symmetric key values."
+ self.assertNotIn('decrypt', key['key_ops'])
+ self.assertNotIn('sign', key['key_ops'])
+ if 'verify' in key['key_ops']:
+ has_sign_keys = True
+ if 'encrypt' in key['key_ops']:
+ has_enc_keys = True
+ if 'alg' in key:
+ self.assertIsInstance(key['alg'], str)
+ if 'kid' in key:
+ self.assertIsInstance(key['kid'], str)
+ self.assertNotIn(key['kid'], kids) # SHOULD
+ kids.add(key['kid'])
+ # ignoring everything X.509 related
+ # TODO: Validate algorithm-specific part of JWK
+ if has_sign_keys and has_encrypt_keys:
+ for key in r_jwks.json['keys']:
+ self.assertIn('use', key)
+
+class TestOIDCBasicProfile(UffdTestCase):
+ def setUpDB(self):
+ db.session.add(OAuth2Key(**TEST_JWK))
+ db.session.add(OAuth2Client(service=Service(name='test', limit_access=False), client_id='test', client_secret='testsecret', redirect_uris=['https://service/callback']))
+
+ # Helper
+ def validate_claim_syntax(self, name, value):
+ # Strip language tag
+ if '#' in name:
+ name = name.split('#')[0]
+ str_claims = ('sub', 'name', 'given_name', 'family_name', 'middle_name', 'nickname', 'preferred_username', 'profile', 'picture', 'website', 'email', 'gender', 'birthdate', 'zoneinfo', 'locale', 'phone_number', 'acr')
+ if name in str_claims:
+ self.assertIsInstance(value, str)
+ if name in ('profile', 'picture', 'website'):
+ self.assertTrue(value.lower().startswith('http'))
+ if name in ('email_verified', 'phone_number_verified'):
+ self.assertIsInstance(value, bool)
+ if name in ('updated_at', 'auth_time'):
+ self.assertTrue(isinstance(value, int) or isinstance(value, float))
+ if name == 'address':
+ self.assertIsInstance(value, dict)
+ if name == 'amr':
+ self.assertIsInstance(value, list)
+ for item in value:
+ self.assertIsInstance(item, str)
+
+ def validate_id_token(self, id_token, nonce='testnonce', client_id='test'):
+ headers = jwt.get_unverified_header(id_token)
+ self.assertIn('kid', headers)
+ self.assertIsInstance(headers['kid'], str)
+ # This checks signature and exp
+ data = OAuth2Key.decode_jwt(id_token, options={'verify_aud': False})
+ self.assertIn('iss', data)
+ self.assertIsInstance(data['iss'], str)
+ self.assertIn('sub', data)
+ self.assertIn('aud', data)
+ self.assertIsInstance(data['aud'], str)
+ if client_id is not None:
+ self.assertEqual(data['aud'], client_id)
+ self.assertIn('iat', data)
+ self.assertTrue(isinstance(data['iat'], int) or isinstance(data['iat'], float))
+ if 'nonce' in data:
+ self.assertIsInstance(data['nonce'], str)
+ self.assertEqual(data.get('nonce'), nonce)
+ if 'azp' in data:
+ self.assertIsInstance(data['azp'], str)
+ for name, value in data.items():
+ self.validate_claim_syntax(name, value)
+ return data
+
+ def is_login_page(self, location):
+ url = urlparse(location)
+ return url.netloc in ('localhost', '') and url.path == url_for('session.login')
+
+ def is_callback(self, location):
+ return location.startswith('https://service/callback')
+
+ def do_auth_request(self, client_id='test', state='teststate', nonce='testnonce', redirect_uri='https://service/callback', scope='openid', follow_redirects=True, **kwargs):
+ r = self.client.get(path=url_for('oauth2.authorize', client_id=client_id, state=state, nonce=nonce, redirect_uri=redirect_uri, scope=scope, **kwargs), follow_redirects=False)
+ while follow_redirects and r.status_code == 302 and not self.is_login_page(r.location) and not self.is_callback(r.location):
+ r = self.client.get(path=r.location, follow_redirects=False)
+ return r
+
+ def do_login(self, r, loginname='testuser', password='userpassword'):
+ self.assertEqual(r.status_code, 302)
+ self.assertTrue(self.is_login_page(r.location))
+ self.client.get(path=url_for('session.logout'), follow_redirects=True)
+ args = parse_qs(urlparse(r.location).query)
+ r = self.client.post(path=url_for('session.login', ref=args['ref'][0]), data={'loginname': loginname, 'password': password}, follow_redirects=False)
+ while r.status_code == 302 and not self.is_login_page(r.location) and not self.is_callback(r.location):
+ r = self.client.get(path=r.location, follow_redirects=False)
+ return r
+
+ def validate_auth_response(self, r, state='teststate'):
+ self.assertEqual(r.status_code, 302)
+ self.assertTrue(self.is_callback(r.location))
+ args = parse_qs(urlparse(r.location).query)
+ for key in args:
+ self.assertNotIn(key, ('error', 'error_description', 'error_uri'))
+ self.assertEqual(len(args[key]), 1) # Not generally specified, but still a good check
+ if state is not None:
+ self.assertIn('state', args)
+ self.assertEqual(args['state'], [state])
+ else:
+ self.assertNotIn('state', args)
+ return {key: values[0] for key, values in args.items()}
+
+ def assert_auth_error(self, r, *errors, state='teststate'):
+ self.assertEqual(r.status_code, 302)
+ self.assertTrue(self.is_callback(r.location))
+ args = parse_qs(urlparse(r.location).query)
+ for key in args:
+ self.assertIn(key, ('error', 'error_description', 'error_uri', 'state'))
+ self.assertEqual(len(args[key]), 1)
+ self.assertIn('error', args)
+ if state is not None:
+ self.assertIn('state', args)
+ self.assertEqual(args['state'][0], state)
+ else:
+ self.assertNotIn('state', args)
+ if errors:
+ self.assertIn(args['error'][0], errors)
+ self.assertRegex(args['error'][0], r'^[ -!#-\[\]-~]+$') # 1*( %x20-x21 / %x23-5B / %x5D-7E )
+ if 'error_description' in args:
+ self.assertRegex(args['error_description'][0], r'^[ -!#-\[\]-~]+$') # 1*( %x20-x21 / %x23-5B / %x5D-7E )
+ if 'error_uri' in args:
+ self.assertRegex(args['error_uri'][0], r'^[!#-\[\]-~]+$') # 1*( %x21 / %x23-5B / %x5D-7E )
+
+ def do_token_request(self, client_id='test', client_secret='testsecret', redirect_uri='https://service/callback', **kwargs):
+ data = {'redirect_uri': redirect_uri, 'client_id': client_id, 'client_secret': client_secret}
+ data.update(kwargs)
+ return self.client.post(path=url_for('oauth2.token'), data=data, follow_redirects=True)
+
+ def validate_token_response(self, r, nonce='testnonce', client_id='test'):
+ self.assertEqual(r.status_code, 200)
+ self.assertEqual(r.content_type, 'application/json')
+ self.assertIn('Cache-Control', r.headers)
+ self.assertEqual(r.headers['Cache-Control'].lower(), 'no-store')
+ for key in r.json:
+ self.assertNotIn(key, ('error', 'error_description', 'error_uri'))
+ self.assertIn('access_token', r.json)
+ self.assertIsInstance(r.json['access_token'], str)
+ self.assertIn('token_type', r.json)
+ self.assertIsInstance(r.json['token_type'], str)
+ # OIDC Core 1.0 section 3.1.3.3:
+ # > The OAuth 2.0 token_type response parameter value MUST be Bearer,
+ # > [...] unless another Token Type has been negotiated with the Client.
+ self.assertEqual(r.json['token_type'].lower(), 'bearer')
+ if 'expires_in' in r.json:
+ self.assertTrue(isinstance(r.json['expires_in'], int) or isinstance(data['expires_in'], float))
+ if 'refresh_token' in r.json:
+ self.assertIsInstance(r.json['refresh_token'], str)
+ if 'scope' in r.json:
+ self.assertIsInstance(r.json['scope'], str)
+ # scope = scope-token *( SP scope-token )
+ # scope-token = 1*( %x21 / %x23-5B / %x5D-7E )
+ self.assertRegex(r.json['scope'], r'^[!#-\[\]-~]+( [!#-\[\]-~]+)*$')
+ # OIDC Core 1.0 section 3.1.3.3:
+ # > All Token Responses that contain tokens, secrets, or other sensitive
+ # > information MUST include the following HTTP response header fields and values:
+ # > Cache-Control: no-store
+ self.assertIn('id_token', r.json)
+ return self.validate_id_token(r.json['id_token'], nonce=nonce, client_id=client_id)
+
+ def assert_token_error(self, r, *errors):
+ self.assertEqual(r.content_type, 'application/json')
+ if r.json.get('error', '') == 'invalid_client':
+ self.assertEqual(r.status_code, 401)
+ else:
+ self.assertEqual(r.status_code, 400)
+ for key in r.json:
+ self.assertIn(key, ('error', 'error_description', 'error_uri'))
+ self.assertIn('error', r.json)
+ if errors:
+ self.assertIn(r.json['error'], errors)
+ self.assertRegex(r.json['error'], r'^[ -!#-\[\]-~]+$') # 1*( %x20-x21 / %x23-5B / %x5D-7E )
+ if 'error_description' in r.json:
+ self.assertRegex(r.json['error_description'], r'^[ -!#-\[\]-~]+$') # 1*( %x20-x21 / %x23-5B / %x5D-7E )
+ if 'error_uri' in r.json:
+ self.assertRegex(r.json['error_uri'], r'^[!#-\[\]-~]+$') # 1*( %x21 / %x23-5B / %x5D-7E )
+
+ def do_userinfo_request(self, access_token):
+ return self.client.get(path=url_for('oauth2.userinfo'), headers=[('Authorization', 'Bearer %s'%access_token)], follow_redirects=True)
+
+ def validate_userinfo_response(self, r):
+ self.assertEqual(r.status_code, 200)
+ # We ignore everything related to encrypted/signed JWT userinfo here
+ self.assertEqual(r.content_type, 'application/json')
+ self.assertIn('sub', r.json)
+ for name, value in r.json.items():
+ self.validate_claim_syntax(name, value)
+
+ def assert_userinfo_error(self, r):
+ self.assertEqual(r.status_code, 401)
+ self.assertEqual(len(r.headers.getlist('WWW-Authenticate')), 1)
+ method, args = (r.headers['WWW-Authenticate'].split(' ', 1) + [''])[:2]
+ args = {item.split('=', 1)[0]: item.split('=', 1)[1].strip(' \n"') for item in args.split(',') if item.strip()}
+ if 'scope' in args:
+ self.assertRegex(args['scope'], r'^[ -!#-\[\]-~]+$') # 1*( %x20-x21 / %x23-5B / %x5D-7E )
+ if 'error' in args:
+ self.assertRegex(args['error'], r'^[ -!#-\[\]-~]+$') # 1*( %x20-x21 / %x23-5B / %x5D-7E )
+ if 'error_description' in args:
+ self.assertRegex(args['error_description'], r'^[ -!#-\[\]-~]+$') # 1*( %x20-x21 / %x23-5B / %x5D-7E )
+
+ def test(self):
+ self.login_as('user')
+ r = self.do_auth_request(response_type='code')
+ args = self.validate_auth_response(r)
+ r = self.do_token_request(grant_type='authorization_code', code=args['code'])
+ id_token = self.validate_token_response(r)
+ self.assertEqual(id_token['sub'], '10000')
+ r = self.do_userinfo_request(r.json['access_token'])
+ self.validate_userinfo_response(r)
+ self.assertEqual(r.json['sub'], '10000')
+
+ def test_notloggedin(self):
+ r = self.do_auth_request(response_type='code')
+ r = self.do_login(r)
+ args = self.validate_auth_response(r)
+ r = self.do_token_request(grant_type='authorization_code', code=args['code'])
+ id_token = self.validate_token_response(r)
+ self.assertEqual(id_token['sub'], '10000')
+
+ def test_no_state(self):
+ self.login_as('user')
+ r = self.do_auth_request(response_type='code', state=None)
+ args = self.validate_auth_response(r, state=None)
+ r = self.do_token_request(grant_type='authorization_code', code=args['code'])
+ id_token = self.validate_token_response(r)
+ self.assertEqual(id_token['sub'], '10000')
+
+ def test_no_nonce(self):
+ self.login_as('user')
+ r = self.do_auth_request(response_type='code', nonce=None)
+ args = self.validate_auth_response(r)
+ r = self.do_token_request(grant_type='authorization_code', code=args['code'])
+ id_token = self.validate_token_response(r, nonce=None)
+ self.assertEqual(id_token['sub'], '10000')
+
+ def test_redirect_uri(self):
+ self.login_as('user')
+ # No redirect_uri in auth request is fine if there is only one uri registered
+ r = self.do_auth_request(response_type='code', redirect_uri=None)
+ args = self.validate_auth_response(r)
+ r = self.do_token_request(grant_type='authorization_code', code=args['code'], redirect_uri=None)
+ id_token = self.validate_token_response(r)
+ self.assertEqual(id_token['sub'], '10000')
+ # If auth request has redirect_uri, it is required in token request
+ r = self.do_auth_request(response_type='code')
+ args = self.validate_auth_response(r)
+ r = self.do_token_request(grant_type='authorization_code', code=args['code'], redirect_uri=None)
+ self.assert_token_error(r)
+ # If auth request has redirect_uri, it the redirect_uri in the token request must be the same
+ r = self.do_auth_request(response_type='code')
+ args = self.validate_auth_response(r)
+ r = self.do_token_request(grant_type='authorization_code', code=args['code'], redirect_uri='https://foobar/callback')
+ self.assert_token_error(r)
+ # Invalid redirect_uri
+ r = self.do_auth_request(response_type='code', redirect_uri='http://foobar/callback')
+ self.assertEqual(r.status_code, 400) # No redirect!
+ # redirect_uri is required in auth request if there is more than a single uri registered
+ client = OAuth2Client.query.one()
+ client.redirect_uris.append('https://service/callback2')
+ db.session.commit()
+ r = self.do_auth_request(response_type='code', redirect_uri=None)
+ self.assertEqual(r.status_code, 400) # No redirect!
+
+ def test_auth_errors(self):
+ # Missing response_type
+ r = self.do_auth_request()
+ self.assert_auth_error(r, 'invalid_request')
+ # Invalid response_type
+ r = self.do_auth_request(response_type='foobar')
+ self.assert_auth_error(r, 'unsupported_response_type')
+ # Missing client_id
+ r = self.do_auth_request(response_type='code', client_id=None)
+ self.assertEqual(r.status_code, 400) # No redirect!
+ # Invalid client_id
+ r = self.do_auth_request(response_type='code', client_id='foobar')
+ self.assertEqual(r.status_code, 400) # No redirect!
+ # Duplicate parameter
+ r = self.do_auth_request(response_type='code', client_id=['test', 'foobar'])
+ self.assertEqual(r.status_code, 400) # No redirect!
+
+ def test_access_denied(self):
+ Service.query.one().limit_access = True
+ db.session.commit()
+ self.login_as('user')
+ r = self.do_auth_request(response_type='code')
+ self.assert_auth_error(r, 'access_denied')
+
+ def test_auth_request_uri(self):
+ self.login_as('user')
+ r = self.do_auth_request(response_type='code', request_uri='https://localhost/myrequest_uri')
+ self.assert_auth_error(r, 'request_uri_not_supported')
+
+ def test_auth_request_unsigned(self):
+ self.login_as('user')
+ request_params = {
+ 'response_type': 'code',
+ 'client_id': 'test',
+ 'redirect_uri': 'http://service/callback',
+ 'scope': 'openid',
+ 'state': 'teststate',
+ 'nonce': 'testnonce',
+ 'claims': {
+ 'userinfo': {
+ 'name': None,
+ 'email': {'essential': True},
+ 'email_verified': {'essential': True},
+ },
+ 'id_token': {
+ 'email': None,
+ }
+ }
+ }
+ r = self.do_auth_request(response_type='code', request=jwt.encode(request_params, algorithm='none', key=None))
+ self.assert_auth_error(r, 'request_not_supported')
+
+ def test_token_client_auth(self):
+ self.login_as('user')
+ # Auth via body -> ACCEPT
+ r = self.do_auth_request(response_type='code')
+ args = self.validate_auth_response(r)
+ r = self.do_token_request(grant_type='authorization_code', code=args['code'])
+ self.validate_token_response(r)
+ # Auth via header -> ACCEPT
+ r = self.do_auth_request(response_type='code')
+ args = self.validate_auth_response(r)
+ r = self.client.post(
+ path=url_for('oauth2.token'),
+ data={'redirect_uri': 'https://service/callback', 'grant_type': 'authorization_code', 'code': args['code']},
+ headers={'Authorization': f'Basic dGVzdDp0ZXN0c2VjcmV0'},
+ follow_redirects=True,
+ )
+ self.validate_token_response(r)
+ # Auth via header, but same client id also in body -> ACCEPT
+ r = self.do_auth_request(response_type='code')
+ args = self.validate_auth_response(r)
+ r = self.client.post(
+ path=url_for('oauth2.token'),
+ data={'redirect_uri': 'https://service/callback', 'grant_type': 'authorization_code', 'client_id': 'test', 'code': args['code']},
+ headers={'Authorization': f'Basic dGVzdDp0ZXN0c2VjcmV0'},
+ follow_redirects=True,
+ )
+ self.validate_token_response(r)
+ # Different client id in body and header -> REJECT
+ r = self.do_auth_request(response_type='code')
+ args = self.validate_auth_response(r)
+ r = self.client.post(
+ path=url_for('oauth2.token'),
+ data={'redirect_uri': 'https://service/callback', 'grant_type': 'authorization_code', 'client_id': 'XXXX', 'code': args['code']},
+ headers={'Authorization': f'Basic dGVzdDp0ZXN0c2VjcmV0'},
+ follow_redirects=True,
+ )
+ self.assert_token_error(r, 'invalid_request')
+ # Duplicate client id in body -> REJECT
+ r = self.do_auth_request(response_type='code')
+ args = self.validate_auth_response(r)
+ r = self.do_token_request(grant_type='authorization_code', code=args['code'], client_id=['test', 'XXXX'])
+ self.assert_token_error(r, 'invalid_request')
+ # Duplicate client secret in body -> REJECT
+ r = self.do_auth_request(response_type='code')
+ args = self.validate_auth_response(r)
+ r = self.do_token_request(grant_type='authorization_code', code=args['code'], client_secret=['testsecret', 'XXXXX'])
+ self.assert_token_error(r, 'invalid_request')
+ # Client secret in body and header -> REJECT
+ r = self.do_auth_request(response_type='code')
+ args = self.validate_auth_response(r)
+ r = self.client.post(
+ path=url_for('oauth2.token'),
+ data={'redirect_uri': 'https://service/callback', 'grant_type': 'authorization_code', 'client_id': 'test', 'client_secret': 'testsecret', 'code': args['code']},
+ headers={'Authorization': f'Basic dGVzdDp0ZXN0c2VjcmV0'},
+ follow_redirects=True,
+ )
+ self.assert_token_error(r, 'invalid_request')
+ # No secret -> REJECT
+ r = self.do_auth_request(response_type='code')
+ args = self.validate_auth_response(r)
+ r = self.do_token_request(grant_type='authorization_code', code=args['code'], client_secret=None)
+ self.assert_token_error(r, 'invalid_client')
+ # No client id but secret -> REJECT
+ r = self.do_auth_request(response_type='code')
+ args = self.validate_auth_response(r)
+ r = self.do_token_request(grant_type='authorization_code', code=args['code'], client_id=None)
+ self.assert_token_error(r, 'invalid_client')
+ # No client id and no secret -> REJECT
+ r = self.do_auth_request(response_type='code')
+ args = self.validate_auth_response(r)
+ r = self.do_token_request(grant_type='authorization_code', code=args['code'], client_id=None, client_secret=None)
+ self.assert_token_error(r, 'invalid_client')
+ # Unknown client id -> REJECT
+ r = self.do_auth_request(response_type='code')
+ args = self.validate_auth_response(r)
+ r = self.do_token_request(grant_type='authorization_code', code=args['code'], client_id='XXXX')
+ self.assert_token_error(r, 'invalid_client')
+
+ def test_token_errors(self):
+ self.login_as('user')
+ # Missing grant_type parameter
+ r = self.do_auth_request(response_type='code')
+ args = self.validate_auth_response(r)
+ r = self.do_token_request(code=args['code'])
+ self.assert_token_error(r, 'invalid_request')
+ # Missing code parameter
+ r = self.do_auth_request(response_type='code')
+ args = self.validate_auth_response(r)
+ r = self.do_token_request(grant_type='authorization_code')
+ self.assert_token_error(r, 'invalid_request')
+ # redirect_uri behaviour is already tested in test_redirect_uri
+ # Invalid grant type
+ r = self.do_auth_request(response_type='code')
+ args = self.validate_auth_response(r)
+ r = self.do_token_request(grant_type='foobar', code=args['code'])
+ self.assert_token_error(r, 'unsupported_grant_type')
+ # Duplicate code parameter
+ r = self.do_auth_request(response_type='code')
+ args = self.validate_auth_response(r)
+ r = self.do_token_request(grant_type='authorization_code', code=[args['code'], 'XXXXX'])
+ self.assert_token_error(r, 'invalid_request')
+ # Invalid code parameter
+ r = self.do_auth_request(response_type='code')
+ args = self.validate_auth_response(r)
+ r = self.do_token_request(grant_type='authorization_code', code='XXXXX')
+ self.assert_token_error(r, 'invalid_grant')
+ # Invalid code parameter
+ r = self.do_auth_request(response_type='code')
+ args = self.validate_auth_response(r)
+ r = self.do_token_request(grant_type='authorization_code', code=args['code'].split('-')[0]+'-XXXXX')
+ self.assert_token_error(r, 'invalid_grant')
+ # Code was issued to different client
+ db.session.add(OAuth2Client(service=Service(name='test2', limit_access=False), client_id='test2', client_secret='testsecret2', redirect_uris=['https://service2/callback']))
+ db.session.commit()
+ r = self.do_auth_request(response_type='code')
+ args = self.validate_auth_response(r)
+ r = self.do_token_request(grant_type='authorization_code', code=args['code'], client_id='test2', client_secret='testsecret2')
+ self.assert_token_error(r, 'invalid_grant')
+
+ def test_userinfo_auth(self):
+ self.login_as('user')
+ r = self.do_auth_request(response_type='code')
+ args = self.validate_auth_response(r)
+ r = self.do_token_request(grant_type='authorization_code', code=args['code'])
+ self.validate_token_response(r)
+ access_token = r.json['access_token']
+ # GET + Bearer
+ r = self.client.get(path=url_for('oauth2.userinfo'), headers=[('Authorization', 'Bearer %s'%access_token)], follow_redirects=True)
+ self.validate_userinfo_response(r)
+ self.assertEqual(r.json['sub'], '10000')
+ # POST + Bearer
+ r = self.client.post(path=url_for('oauth2.userinfo'), headers=[('Authorization', 'Bearer %s'%access_token)], follow_redirects=True)
+ self.validate_userinfo_response(r)
+ self.assertEqual(r.json['sub'], '10000')
+ # "Bearer" is case-insensitive
+ r = self.client.post(path=url_for('oauth2.userinfo'), headers=[('authorization', 'bearer %s'%access_token)], follow_redirects=True)
+ self.validate_userinfo_response(r)
+ self.assertEqual(r.json['sub'], '10000')
+ # Invalid auth scheme
+ r = self.client.post(path=url_for('oauth2.userinfo'), headers=[('Authorization', 'Basic dGVzdDp0ZXN0c2VjcmV0')], follow_redirects=True)
+ self.assert_userinfo_error(r)
+ # Invalid bearer token
+ r = self.client.post(path=url_for('oauth2.userinfo'), headers=[('Authorization', 'Bearer %s-XXXXX'%access_token.split('-')[0])], follow_redirects=True)
+ self.assert_userinfo_error(r)
+ r = self.client.post(path=url_for('oauth2.userinfo'), headers=[('Authorization', 'Bearer XXXXX')], follow_redirects=True)
+ self.assert_userinfo_error(r)
+ # POST + body
+ r = self.client.post(path=url_for('oauth2.userinfo'), data={'access_token': access_token}, follow_redirects=True)
+ self.validate_userinfo_response(r)
+ self.assertEqual(r.json['sub'], '10000')
+ # GET + query
+ r = self.client.get(path=url_for('oauth2.userinfo', access_token=access_token), follow_redirects=True)
+ self.validate_userinfo_response(r)
+ self.assertEqual(r.json['sub'], '10000')
+ # POST + Bearer + body -> REJECT
+ r = self.client.post(path=url_for('oauth2.userinfo'), data={'access_token': access_token}, headers=[('Authorization', 'Bearer %s'%access_token)], follow_redirects=True)
+ self.assert_userinfo_error(r)
+ # No auth -> REJECT
+ r = self.client.post(path=url_for('oauth2.userinfo'), follow_redirects=True)
+ self.assert_userinfo_error(r)
+
+ def test_scope(self):
+ self.login_as('user')
+ # Scope values used that are not understood by an implementation SHOULD be ignored.
+ r = self.do_auth_request(response_type='code', scope='openid profile email address phone groups foobar')
+ args = self.validate_auth_response(r)
+ r = self.do_token_request(grant_type='authorization_code', code=args['code'])
+ id_token = self.validate_token_response(r)
+ self.assertEqual(id_token['sub'], '10000')
+ r = self.do_userinfo_request(r.json['access_token'])
+ self.validate_userinfo_response(r)
+ self.assertEqual(r.json['sub'], '10000')
+ self.assertEqual(r.json['name'], 'Test User')
+ self.assertEqual(r.json['preferred_username'], 'testuser')
+ self.assertEqual(r.json['email'], 'test@example.com')
+ self.assertEqual(sorted(r.json['groups']), sorted(['users', 'uffd_access']))
+
+ def test_claims(self):
+ self.login_as('user')
+ # Scope values used that are not understood by an implementation SHOULD be ignored.
+ r = self.do_auth_request(response_type='code', claims='{"userinfo": {"name": {"essential": true}}, "id_token": {"preferred_username": {"essential": true}, "email": null}}')
+ args = self.validate_auth_response(r)
+ r = self.do_token_request(grant_type='authorization_code', code=args['code'])
+ id_token = self.validate_token_response(r)
+ self.assertEqual(id_token['sub'], '10000')
+ self.assertEqual(id_token['preferred_username'], 'testuser')
+ self.assertEqual(id_token['email'], 'test@example.com')
+ self.assertNotIn('name', r.json)
+ r = self.do_userinfo_request(r.json['access_token'])
+ self.validate_userinfo_response(r)
+ self.assertEqual(r.json['sub'], '10000')
+ self.assertEqual(r.json['name'], 'Test User')
+ self.assertNotIn('email', r.json)
+
+ def test_prompt_none(self):
+ r = self.do_auth_request(response_type='code', prompt='none')
+ self.assert_auth_error(r, 'login_required')
+ self.login_as('user')
+ r = self.do_auth_request(response_type='code', prompt='none')
+ args = self.validate_auth_response(r)
+ self.assertIn('code', args)
+ # OIDC Core 1.0 section 3.1.2.1.:
+ # > If this parameter contains none with any other value, an error is returned.
+ r = self.do_auth_request(response_type='code', prompt='none login')
+ self.assert_auth_error(r)
+
+ @unittest.skip('prompt=login is not implemented') # MUST
+ def test_prompt_login(self):
+ self.login_as('user')
+ r = self.do_auth_request(response_type='code')
+ args = self.validate_auth_response(r)
+ self.assertIn('code', args)
+ r = self.do_auth_request(response_type='code', prompt='login')
+ self.assertEqual(r.status_code, 302)
+ self.assertTrue(self.is_login_page(r.location))
+
+ # TODO: max_age
+
+ def test_sub_value(self):
+ # Via id_token_hint or claims.id_token.sub.value
+ self.login_as('user')
+ r = self.do_auth_request(response_type='code', prompt='none')
+ args = self.validate_auth_response(r)
+ r = self.do_token_request(grant_type='authorization_code', code=args['code'])
+ self.validate_token_response(r)
+ id_token = r.json['id_token']
+ r = self.do_auth_request(response_type='code', prompt='none', id_token_hint=id_token)
+ args = self.validate_auth_response(r)
+ self.assertIn('code', args)
+ r = self.do_auth_request(response_type='code', prompt='none', id_token_hint='XXXXX')
+ self.assert_auth_error(r, 'invalid_request')
+ r = self.do_auth_request(response_type='code', prompt='none', claims='{"id_token": {"sub": {"value": "10000"}}}')
+ args = self.validate_auth_response(r)
+ r = self.do_auth_request(response_type='code', prompt='none', claims='{"id_token": {"sub": {"value": "10001"}}}')
+ self.assert_auth_error(r, 'login_required')
+ # sub value in id_token_hint and claims is the same -> Not ambiguous
+ r = self.do_auth_request(response_type='code', prompt='none', id_token_hint=id_token, claims='{"id_token": {"sub": {"value": "10000"}}}')
+ args = self.validate_auth_response(r)
+ self.assertIn('code', args)
+ # sub value in id_token_hint and claims differs -> Ambiguous
+ r = self.do_auth_request(response_type='code', prompt='none', id_token_hint=id_token, claims='{"id_token": {"sub": {"value": "10001"}}}')
+ self.assert_auth_error(r, 'invalid_request')
+ self.login_as('admin')
+ r = self.do_auth_request(response_type='code', prompt='none', id_token_hint=id_token)
+ self.assert_auth_error(r, 'login_required')
+
+ def test_code_reuse(self):
+ self.login_as('user')
+ r = self.do_auth_request(response_type='code')
+ args = self.validate_auth_response(r)
+ r1 = self.do_token_request(grant_type='authorization_code', code=args['code'])
+ self.validate_token_response(r1)
+ r2 = self.do_token_request(grant_type='authorization_code', code=args['code'])
+ self.assert_token_error(r2, 'invalid_grant')
+
+ @unittest.skip('Token revoking on reuse is not implemented') # SHOULD
+ def test_code_reuse_revoke(self):
+ self.login_as('user')
+ r = self.do_auth_request(response_type='code')
+ args = self.validate_auth_response(r)
+ r1 = self.do_token_request(grant_type='authorization_code', code=args['code'])
+ self.validate_token_response(r1)
+ r2 = self.do_token_request(grant_type='authorization_code', code=args['code'])
+ self.assert_token_error(r2, 'invalid_grant')
+ r = self.do_userinfo_request(r1.json['access_token'])
+ self.assert_userinfo_error(r)
diff --git a/uffd/migrations/versions/01fdd7820f29_openid_connect_support.py b/uffd/migrations/versions/01fdd7820f29_openid_connect_support.py
new file mode 100644
index 00000000..c7a97b2f
--- /dev/null
+++ b/uffd/migrations/versions/01fdd7820f29_openid_connect_support.py
@@ -0,0 +1,148 @@
+"""OpenID Connect Support
+
+Revision ID: 01fdd7820f29
+Revises: a9b449776953
+Create Date: 2023-11-09 16:52:20.860871
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+import datetime
+import secrets
+import math
+import logging
+
+from cryptography.hazmat.primitives.asymmetric import rsa
+from cryptography.hazmat.backends import default_backend # Only required for Buster
+import jwt
+
+# pyjwt v1.7.x compat (Buster/Bullseye)
+if not hasattr(jwt, 'get_algorithm_by_name'):
+ jwt.get_algorithm_by_name = lambda name: jwt.algorithms.get_default_algorithms()[name]
+
+# revision identifiers, used by Alembic.
+revision = '01fdd7820f29'
+down_revision = 'a9b449776953'
+branch_labels = None
+depends_on = None
+
+logger = logging.getLogger('alembic.runtime.migration.01fdd7820f29')
+
+def token_with_alphabet(alphabet, nbytes=None):
+ '''Return random text token that consists of characters from `alphabet`'''
+ if nbytes is None:
+ nbytes = max(secrets.DEFAULT_ENTROPY, 32)
+ nbytes_per_char = math.log(len(alphabet), 256)
+ nchars = math.ceil(nbytes / nbytes_per_char)
+ return ''.join([secrets.choice(alphabet) for _ in range(nchars)])
+
+def token_urlfriendly(nbytes=None):
+ '''Return random text token that is urlsafe and works around common parsing bugs'''
+ alphabet = '0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ'
+ return token_with_alphabet(alphabet, nbytes=nbytes)
+
+def upgrade():
+ logger.info('Generating 3072 bit RSA key pair (RS256) for OpenID Connect support ...')
+ private_key = rsa.generate_private_key(public_exponent=65537, key_size=3072, backend=default_backend())
+
+ meta = sa.MetaData(bind=op.get_bind())
+ oauth2_key = op.create_table('oauth2_key',
+ sa.Column('id', sa.String(length=64), nullable=False),
+ sa.Column('created', sa.DateTime(), nullable=False),
+ sa.Column('active', sa.Boolean(create_constraint=False), nullable=False),
+ sa.Column('algorithm', sa.String(length=32), nullable=False),
+ sa.Column('private_key_jwk', sa.Text(), nullable=False),
+ sa.Column('public_key_jwk', sa.Text(), nullable=False),
+ sa.PrimaryKeyConstraint('id', name=op.f('pk_oauth2_key'))
+ )
+ algorithm = jwt.get_algorithm_by_name('RS256')
+ op.bulk_insert(oauth2_key, [{
+ 'id': token_urlfriendly(),
+ 'created': datetime.datetime.utcnow(),
+ 'active': True,
+ 'algorithm': 'RS256',
+ 'private_key_jwk': algorithm.to_jwk(private_key),
+ 'public_key_jwk': algorithm.to_jwk(private_key.public_key()),
+ }])
+
+ with op.batch_alter_table('oauth2grant', schema=None) as batch_op:
+ batch_op.drop_index(batch_op.f('ix_oauth2grant_code'))
+ oauth2grant = sa.Table('oauth2grant', meta,
+ sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
+ sa.Column('user_id', sa.Integer(), nullable=False),
+ sa.Column('client_db_id', sa.Integer(), nullable=False),
+ sa.Column('code', sa.String(length=255), nullable=False),
+ sa.Column('redirect_uri', sa.String(length=255), nullable=False),
+ sa.Column('expires', sa.DateTime(), nullable=False),
+ sa.Column('_scopes', sa.Text(), nullable=False),
+ sa.ForeignKeyConstraint(['client_db_id'], ['oauth2client.db_id'], name=op.f('fk_oauth2grant_client_db_id_oauth2client'), onupdate='CASCADE', ondelete='CASCADE'),
+ sa.ForeignKeyConstraint(['user_id'], ['user.id'], name=op.f('fk_oauth2grant_user_id_user'), onupdate='CASCADE', ondelete='CASCADE'),
+ sa.PrimaryKeyConstraint('id', name=op.f('pk_oauth2grant'))
+ )
+ with op.batch_alter_table('oauth2grant', copy_from=oauth2grant) as batch_op:
+ batch_op.add_column(sa.Column('nonce', sa.Text(), nullable=True))
+ batch_op.add_column(sa.Column('claims', sa.Text(), nullable=True))
+ batch_op.alter_column('redirect_uri', existing_type=sa.VARCHAR(length=255), nullable=True)
+
+ oauth2token = sa.Table('oauth2token', meta,
+ sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
+ sa.Column('user_id', sa.Integer(), nullable=False),
+ sa.Column('client_db_id', sa.Integer(), nullable=False),
+ sa.Column('token_type', sa.String(length=40), nullable=False),
+ sa.Column('access_token', sa.String(length=255), nullable=False),
+ sa.Column('refresh_token', sa.String(length=255), nullable=False),
+ sa.Column('expires', sa.DateTime(), nullable=False),
+ sa.Column('_scopes', sa.Text(), nullable=False),
+ sa.ForeignKeyConstraint(['client_db_id'], ['oauth2client.db_id'], name=op.f('fk_oauth2token_client_db_id_oauth2client'), onupdate='CASCADE', ondelete='CASCADE'),
+ sa.ForeignKeyConstraint(['user_id'], ['user.id'], name=op.f('fk_oauth2token_user_id_user'), onupdate='CASCADE', ondelete='CASCADE'),
+ sa.PrimaryKeyConstraint('id', name=op.f('pk_oauth2token')),
+ sa.UniqueConstraint('access_token', name=op.f('uq_oauth2token_access_token')),
+ sa.UniqueConstraint('refresh_token', name=op.f('uq_oauth2token_refresh_token'))
+ )
+ with op.batch_alter_table('oauth2token', copy_from=oauth2token) as batch_op:
+ batch_op.add_column(sa.Column('claims', sa.Text(), nullable=True))
+
+def downgrade():
+ meta = sa.MetaData(bind=op.get_bind())
+
+ oauth2token = sa.Table('oauth2token', meta,
+ sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
+ sa.Column('user_id', sa.Integer(), nullable=False),
+ sa.Column('client_db_id', sa.Integer(), nullable=False),
+ sa.Column('token_type', sa.String(length=40), nullable=False),
+ sa.Column('access_token', sa.String(length=255), nullable=False),
+ sa.Column('refresh_token', sa.String(length=255), nullable=False),
+ sa.Column('expires', sa.DateTime(), nullable=False),
+ sa.Column('_scopes', sa.Text(), nullable=False),
+ sa.Column('claims', sa.Text(), nullable=True),
+ sa.ForeignKeyConstraint(['client_db_id'], ['oauth2client.db_id'], name=op.f('fk_oauth2token_client_db_id_oauth2client'), onupdate='CASCADE', ondelete='CASCADE'),
+ sa.ForeignKeyConstraint(['user_id'], ['user.id'], name=op.f('fk_oauth2token_user_id_user'), onupdate='CASCADE', ondelete='CASCADE'),
+ sa.PrimaryKeyConstraint('id', name=op.f('pk_oauth2token')),
+ sa.UniqueConstraint('access_token', name=op.f('uq_oauth2token_access_token')),
+ sa.UniqueConstraint('refresh_token', name=op.f('uq_oauth2token_refresh_token'))
+ )
+ with op.batch_alter_table('oauth2token', copy_from=oauth2token) as batch_op:
+ batch_op.drop_column('claims')
+
+ oauth2grant = sa.Table('oauth2grant', meta,
+ sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
+ sa.Column('user_id', sa.Integer(), nullable=False),
+ sa.Column('client_db_id', sa.Integer(), nullable=False),
+ sa.Column('code', sa.String(length=255), nullable=False),
+ sa.Column('redirect_uri', sa.String(length=255), nullable=True),
+ sa.Column('nonce', sa.Text(), nullable=True),
+ sa.Column('expires', sa.DateTime(), nullable=False),
+ sa.Column('_scopes', sa.Text(), nullable=False),
+ sa.Column('claims', sa.Text(), nullable=True),
+ sa.ForeignKeyConstraint(['client_db_id'], ['oauth2client.db_id'], name=op.f('fk_oauth2grant_client_db_id_oauth2client'), onupdate='CASCADE', ondelete='CASCADE'),
+ sa.ForeignKeyConstraint(['user_id'], ['user.id'], name=op.f('fk_oauth2grant_user_id_user'), onupdate='CASCADE', ondelete='CASCADE'),
+ sa.PrimaryKeyConstraint('id', name=op.f('pk_oauth2grant'))
+ )
+ with op.batch_alter_table('oauth2grant', copy_from=oauth2grant) as batch_op:
+ batch_op.alter_column('redirect_uri', existing_type=sa.VARCHAR(length=255), nullable=False)
+ batch_op.drop_column('claims')
+ batch_op.drop_column('nonce')
+ batch_op.create_index(batch_op.f('ix_oauth2grant_code'), ['code'], unique=False)
+
+ op.drop_table('oauth2_key')
diff --git a/uffd/models/__init__.py b/uffd/models/__init__.py
index 5fe24173..756a9721 100644
--- a/uffd/models/__init__.py
+++ b/uffd/models/__init__.py
@@ -2,7 +2,7 @@ from .api import APIClient
from .invite import Invite, InviteGrant, InviteSignup
from .mail import Mail, MailReceiveAddress, MailDestinationAddress
from .mfa import MFAType, MFAMethod, RecoveryCodeMethod, TOTPMethod, WebauthnMethod
-from .oauth2 import OAuth2Client, OAuth2RedirectURI, OAuth2LogoutURI, OAuth2Grant, OAuth2Token, OAuth2DeviceLoginInitiation
+from .oauth2 import OAuth2Client, OAuth2RedirectURI, OAuth2LogoutURI, OAuth2Grant, OAuth2Token, OAuth2DeviceLoginInitiation, OAuth2Key
from .role import Role, RoleGroup, RoleGroupMap
from .selfservice import PasswordToken
from .service import RemailerMode, Service, ServiceUser, get_services
diff --git a/uffd/models/oauth2.py b/uffd/models/oauth2.py
index b79dd792..e8fd0cdd 100644
--- a/uffd/models/oauth2.py
+++ b/uffd/models/oauth2.py
@@ -1,17 +1,25 @@
import datetime
import json
+import secrets
+import base64
-from sqlalchemy import Column, Integer, String, DateTime, Text, ForeignKey
+from sqlalchemy import Column, Integer, String, DateTime, Text, ForeignKey, Boolean
from sqlalchemy.orm import relationship
from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy.ext.associationproxy import association_proxy
+import jwt
from uffd.database import db, CommaSeparatedList
from uffd.tasks import cleanup_task
from uffd.password_hash import PasswordHashAttribute, HighEntropyPasswordHash
+from uffd.utils import token_urlfriendly
from .session import DeviceLoginInitiation, DeviceLoginType
from .service import ServiceUser
+# pyjwt v1.7.x compat (Buster/Bullseye)
+if not hasattr(jwt, 'get_algorithm_by_name'):
+ jwt.get_algorithm_by_name = lambda name: jwt.algorithms.get_default_algorithms()[name]
+
class OAuth2Client(db.Model):
__tablename__ = 'oauth2client'
# Inconsistently named "db_id" instead of "id" because of the naming conflict
@@ -28,17 +36,9 @@ class OAuth2Client(db.Model):
redirect_uris = association_proxy('_redirect_uris', 'uri')
logout_uris = relationship('OAuth2LogoutURI', cascade='all, delete-orphan')
- @property
- def client_type(self):
- return 'confidential'
-
- @property
- def default_scopes(self):
- return ['profile']
-
@property
def default_redirect_uri(self):
- return self.redirect_uris[0]
+ return self.redirect_uris[0] if len(self.redirect_uris) == 1 else None
def access_allowed(self, user):
service_user = ServiceUser.query.get((self.service_id, user.id))
@@ -69,28 +69,74 @@ class OAuth2Grant(db.Model):
__tablename__ = 'oauth2grant'
id = Column(Integer, primary_key=True, autoincrement=True)
+ EXPIRES_IN = 100
+ expires = Column(DateTime, nullable=False, default=lambda: datetime.datetime.utcnow() + datetime.timedelta(seconds=OAuth2Grant.EXPIRES_IN))
+
user_id = Column(Integer(), ForeignKey('user.id', onupdate='CASCADE', ondelete='CASCADE'), nullable=False)
user = relationship('User')
client_db_id = Column(Integer, ForeignKey('oauth2client.db_id', onupdate='CASCADE', ondelete='CASCADE'), nullable=False)
client = relationship('OAuth2Client')
- code = Column(String(255), index=True, nullable=False)
- redirect_uri = Column(String(255), nullable=False)
- expires = Column(DateTime, nullable=False, default=lambda: datetime.datetime.utcnow() + datetime.timedelta(seconds=100))
+ _code = Column('code', String(255), nullable=False, default=token_urlfriendly)
+ code = property(lambda self: f'{self.id}-{self._code}')
+ redirect_uri = Column(String(255), nullable=True)
+ nonce = Column(Text(), nullable=True)
scopes = Column('_scopes', CommaSeparatedList(), nullable=False, default=tuple())
+ _claims = Column('claims', Text(), nullable=True)
+
+ @property
+ def claims(self):
+ return json.loads(self._claims) if self._claims is not None else None
+
+ @claims.setter
+ def claims(self, value):
+ self._claims = json.dumps(value) if value is not None else None
+
+ @property
+ def service_user(self):
+ service_user = ServiceUser.query.get((self.client.service_id, self.user.id))
+ if service_user is None:
+ raise Exception('ServiceUser lookup failed')
+ return service_user
+
@hybrid_property
def expired(self):
if self.expires is None:
return False
return self.expires < datetime.datetime.utcnow()
+ @classmethod
+ def get_by_authorization_code(cls, code):
+ # pylint: disable=protected-access
+ if '-' not in code:
+ return None
+ grant_id, grant_code = code.split('-', 2)
+ grant = cls.query.filter_by(id=grant_id, expired=False).first()
+ if not grant or not secrets.compare_digest(grant._code, grant_code):
+ return None
+ if grant.user.is_deactivated or not grant.client.access_allowed(grant.user):
+ return None
+ return grant
+
+ def make_token(self, **kwargs):
+ return OAuth2Token(
+ user=self.user,
+ client=self.client,
+ scopes=self.scopes,
+ claims=self.claims,
+ **kwargs
+ )
+
@cleanup_task.delete_by_attribute('expired')
class OAuth2Token(db.Model):
__tablename__ = 'oauth2token'
id = Column(Integer, primary_key=True, autoincrement=True)
+ EXPIRES_IN = 3600
+ expires = Column(DateTime, nullable=False, default=lambda: datetime.datetime.utcnow() + datetime.timedelta(seconds=OAuth2Token.EXPIRES_IN))
+
user_id = Column(Integer(), ForeignKey('user.id', onupdate='CASCADE', ondelete='CASCADE'), nullable=False)
user = relationship('User')
@@ -98,19 +144,46 @@ class OAuth2Token(db.Model):
client = relationship('OAuth2Client')
# currently only bearer is supported
- token_type = Column(String(40), nullable=False)
- access_token = Column(String(255), unique=True, nullable=False)
- refresh_token = Column(String(255), unique=True, nullable=False)
- expires = Column(DateTime, nullable=False)
+ token_type = Column(String(40), nullable=False, default='bearer')
+ _access_token = Column('access_token', String(255), unique=True, nullable=False, default=token_urlfriendly)
+ access_token = property(lambda self: f'{self.id}-{self._access_token}')
+ _refresh_token = Column('refresh_token', String(255), unique=True, nullable=False, default=token_urlfriendly)
+ refresh_token = property(lambda self: f'{self.id}-{self._refresh_token}')
scopes = Column('_scopes', CommaSeparatedList(), nullable=False, default=tuple())
+ _claims = Column('claims', Text(), nullable=True)
+
+ @property
+ def claims(self):
+ return json.loads(self._claims) if self._claims is not None else None
+
+ @claims.setter
+ def claims(self, value):
+ self._claims = json.dumps(value) if value is not None else None
+
+ @property
+ def service_user(self):
+ service_user = ServiceUser.query.get((self.client.service_id, self.user.id))
+ if service_user is None:
+ raise Exception('ServiceUser lookup failed')
+ return service_user
+
@hybrid_property
def expired(self):
return self.expires < datetime.datetime.utcnow()
- def set_expires_in_seconds(self, seconds):
- self.expires = datetime.datetime.utcnow() + datetime.timedelta(seconds=seconds)
- expires_in_seconds = property(fset=set_expires_in_seconds)
+ @classmethod
+ def get_by_access_token(cls, access_token):
+ # pylint: disable=protected-access
+ if '-' not in access_token:
+ return None
+ token_id, token_secret = access_token.split('-', 2)
+ token = cls.query.filter_by(id=token_id, expired=False).first()
+ if not token or not secrets.compare_digest(token._access_token, token_secret):
+ return None
+ if token.user.is_deactivated or not token.client.access_allowed(token.user):
+ return None
+ return token
class OAuth2DeviceLoginInitiation(DeviceLoginInitiation):
__mapper_args__ = {
@@ -122,3 +195,93 @@ class OAuth2DeviceLoginInitiation(DeviceLoginInitiation):
@property
def description(self):
return self.client.service.name
+
+class OAuth2Key(db.Model):
+ __tablename__ = 'oauth2_key'
+ id = Column(String(64), primary_key=True, default=token_urlfriendly)
+ created = Column(DateTime, default=datetime.datetime.utcnow, nullable=False)
+ active = Column(Boolean(create_constraint=False), default=True, nullable=False)
+ algorithm = Column(String(32), nullable=False)
+ private_key_jwk = Column(Text(), nullable=False)
+ public_key_jwk = Column(Text(), nullable=False)
+
+ def __init__(self, **kwargs):
+ if kwargs.get('algorithm') and kwargs.get('private_key') \
+ and not kwargs.get('private_key_jwk') \
+ and not kwargs.get('public_key_jwk'):
+ algorithm = jwt.get_algorithm_by_name(kwargs['algorithm'])
+ private_key = kwargs.pop('private_key')
+ kwargs['private_key_jwk'] = algorithm.to_jwk(private_key)
+ kwargs['public_key_jwk'] = algorithm.to_jwk(private_key.public_key())
+ super().__init__(**kwargs)
+
+ @property
+ def private_key(self):
+ # pylint: disable=protected-access,import-outside-toplevel
+ # cryptography performs expensive checks when loading RSA private keys.
+ # Since we only load keys we generated ourselves with help of cryptography,
+ # these checks are unnecessary.
+ import cryptography.hazmat.backends.openssl
+ cryptography.hazmat.backends.openssl.backend._rsa_skip_check_key = True
+ res = jwt.get_algorithm_by_name(self.algorithm).from_jwk(self.private_key_jwk)
+ cryptography.hazmat.backends.openssl.backend._rsa_skip_check_key = False
+ return res
+
+ @property
+ def public_key(self):
+ return jwt.get_algorithm_by_name(self.algorithm).from_jwk(self.public_key_jwk)
+
+ @property
+ def public_key_jwks_dict(self):
+ res = json.loads(self.public_key_jwk)
+ res['kid'] = self.id
+ res['alg'] = self.algorithm
+ res['use'] = 'sig'
+ # RFC7517 4.3 "The "use" and "key_ops" JWK members SHOULD NOT be used together [...]"
+ res.pop('key_ops', None)
+ return res
+
+ def encode_jwt(self, payload):
+ if not self.active:
+ raise jwt.exceptions.InvalidKeyError(f'Key {self.id} not active')
+ return jwt.encode(payload, key=self.private_key, algorithm=self.algorithm, headers={'kid': self.id})
+
+ # Hash algorithm for at_hash/c_hash from OpenID Connect Core 1.0 section 3.1.3.6
+ def oidc_hash(self, value):
+ # pylint: disable=import-outside-toplevel
+ from cryptography.hazmat.primitives import hashes
+ from cryptography.hazmat.backends import default_backend # Only required for Buster
+ hash_alg = jwt.get_algorithm_by_name(self.algorithm).hash_alg
+ digest = hashes.Hash(hash_alg(), backend=default_backend())
+ digest.update(value)
+ return base64.urlsafe_b64encode(
+ digest.finalize()[:hash_alg.digest_size // 2]
+ ).decode('ascii').rstrip('=')
+
+ @classmethod
+ def get_preferred_key(cls, algorithm='RS256'):
+ return cls.query.filter_by(active=True, algorithm=algorithm).order_by(OAuth2Key.created.desc()).first()
+
+ @classmethod
+ def get_available_algorithms(cls):
+ return ['RS256']
+
+ @classmethod
+ def decode_jwt(cls, data, algorithms=('RS256',), **kwargs):
+ headers = jwt.get_unverified_header(data)
+ if 'kid' not in headers:
+ raise jwt.exceptions.InvalidKeyError('JWT without kid')
+ kid = headers['kid']
+ key = cls.query.get(kid)
+ if not key:
+ raise jwt.exceptions.InvalidKeyError(f'Key {kid} not found')
+ if not key.active:
+ raise jwt.exceptions.InvalidKeyError(f'Key {kid} not active')
+ return jwt.decode(data, key=key.public_key, algorithms=algorithms, **kwargs)
+
+ @classmethod
+ def generate_rsa_key(cls, public_exponent=65537, key_size=3072):
+ # pylint: disable=import-outside-toplevel
+ from cryptography.hazmat.primitives.asymmetric import rsa
+ from cryptography.hazmat.backends import default_backend # Only required for Buster
+ return cls(algorithm='RS256', private_key=rsa.generate_private_key(public_exponent=public_exponent, key_size=key_size, backend=default_backend()))
diff --git a/uffd/models/service.py b/uffd/models/service.py
index 1bed5cfc..c37f6b37 100644
--- a/uffd/models/service.py
+++ b/uffd/models/service.py
@@ -116,6 +116,16 @@ class ServiceUser(db.Model):
return remailer.build_v2_address(self.service_id, self.user_id)
return self.real_email
+ # User.primary_email and ServiceUser.service_email can only be set to
+ # verified addresses, so this should always return True
+ @property
+ def email_verified(self):
+ if self.effective_remailer_mode != RemailerMode.DISABLED:
+ return True
+ if self.has_email_preferences and self.service_email:
+ return self.service_email.verified
+ return self.user.primary_email.verified
+
@classmethod
def filter_query_by_email(cls, query, email):
'''Filter query of ServiceUser by ServiceUser.email'''
diff --git a/uffd/views/oauth2.py b/uffd/views/oauth2.py
index becb485a..f46305a1 100644
--- a/uffd/views/oauth2.py
+++ b/uffd/views/oauth2.py
@@ -1,253 +1,490 @@
-import functools
-import secrets
import urllib.parse
+import time
+import json
from flask import Blueprint, request, jsonify, render_template, session, redirect, url_for, flash, abort
-import oauthlib.oauth2
from flask_babel import gettext as _
from sqlalchemy.exc import IntegrityError
+import jwt
from uffd.secure_redirect import secure_local_redirect
from uffd.database import db
-from uffd.models import DeviceLoginConfirmation, OAuth2Client, OAuth2Grant, OAuth2Token, OAuth2DeviceLoginInitiation, host_ratelimit, format_delay, ServiceUser
-
-class UffdRequestValidator(oauthlib.oauth2.RequestValidator):
- # Argument "oauthreq" is named "request" in superclass but this clashes with flask's "request" object
- # Arguments "token_value" and "token_data" are named "token" in superclass but this clashs with "token" endpoint
- # pylint: disable=arguments-differ,arguments-renamed,unused-argument,too-many-public-methods,abstract-method
-
- # In all cases (aside from validate_bearer_token), either validate_client_id or authenticate_client is called
- # before anything else. authenticate_client_id would be called instead of authenticate_client for non-confidential
- # clients. However, we don't support those.
- def validate_client_id(self, client_id, oauthreq, *args, **kwargs):
- oauthreq.client = OAuth2Client.query.filter_by(client_id=client_id).one_or_none()
- return oauthreq.client is not None
-
- def authenticate_client(self, oauthreq, *args, **kwargs):
- authorization = oauthreq.extra_credentials.get('authorization')
- if authorization:
- # From RFC6749 2.3.1:
- # Clients in possession of a client password MAY use the HTTP Basic authentication
- # scheme as defined in [RFC2617] to authenticate with the authorization server.
- # The client identifier is encoded using the "application/x-www-form-urlencoded"
- # encoding algorithm per Appendix B, and the encoded value is used as the username
- # the client password is encoded using the same algorithm and used as the password.
- oauthreq.client_id = urllib.parse.unquote(authorization.username)
- oauthreq.client_secret = urllib.parse.unquote(authorization.password)
- if oauthreq.client_secret is None:
- return False
- oauthreq.client = OAuth2Client.query.filter_by(client_id=oauthreq.client_id).one_or_none()
- if oauthreq.client is None:
- return False
- if not oauthreq.client.client_secret.verify(oauthreq.client_secret):
- return False
- if oauthreq.client.client_secret.needs_rehash:
- oauthreq.client.client_secret = oauthreq.client_secret
- db.session.commit()
- return True
+from uffd.models import (
+ DeviceLoginConfirmation, OAuth2Client, OAuth2Grant, OAuth2Token, OAuth2DeviceLoginInitiation,
+ host_ratelimit, format_delay, OAuth2Key,
+)
- def get_default_redirect_uri(self, client_id, oauthreq, *args, **kwargs):
- return oauthreq.client.default_redirect_uri
+def get_issuer():
+ return request.host_url.rstrip('/')
- def validate_redirect_uri(self, client_id, redirect_uri, oauthreq, *args, **kwargs):
- return redirect_uri in oauthreq.client.redirect_uris
+OIDC_SCOPES = {
+ # From https://openid.net/specs/openid-connect-core-1_0.html
+ 'openid': {
+ # "The sub (subject) Claim MUST always be returned in the UserInfo Response."
+ 'sub': None,
+ },
+ 'profile': {
+ 'name': None,
+ 'family_name': None,
+ 'given_name': None,
+ 'middle_name': None,
+ 'nickname': None,
+ 'preferred_username': None,
+ 'profile': None,
+ 'picture': None,
+ 'website': None,
+ 'gender': None,
+ 'birthdate': None,
+ 'zoneinfo': None,
+ 'locale': None,
+ 'updated_at': None,
+ },
+ 'email': {
+ 'email': None,
+ 'email_verified': None,
+ },
+ # Custom scopes
+ 'groups': {
+ 'groups': None,
+ },
+}
- def validate_response_type(self, client_id, response_type, client, oauthreq, *args, **kwargs):
- return response_type == 'code'
+OIDC_CLAIMS = {
+ 'sub': lambda service_user: str(service_user.user.unix_uid),
+ 'name': lambda service_user: service_user.user.displayname,
+ 'preferred_username': lambda service_user: service_user.user.loginname,
+ 'email': lambda service_user: service_user.email,
+ 'email_verified': lambda service_user: service_user.email_verified,
+ # RFC 9068 registers the "groups" claim with a syntax taken from SCIM (RFC 7643)
+ # that is different from what we use here. The plain list of names syntax we use
+ # is far more common in the context of id_token/userinfo claims.
+ 'groups': lambda service_user: [group.name for group in service_user.user.groups]
+}
- def get_default_scopes(self, client_id, oauthreq, *args, **kwargs):
- return oauthreq.client.default_scopes
+def render_claims(scopes, claims, service_user):
+ claims = dict(claims)
+ for scope in scopes:
+ claims.update(OIDC_SCOPES.get(scope, {}))
+ # This would be a good place to enforce permissions on available claims
+ res = {}
+ for claim, func in OIDC_CLAIMS.items():
+ if claim in claims:
+ res[claim] = func(service_user=service_user)
+ return res
- def validate_scopes(self, client_id, scopes, client, oauthreq, *args, **kwargs):
- if scopes == ['']:
- oauthreq.scopes = scopes = self.get_default_scopes(client_id, oauthreq)
- return set(scopes).issubset({'profile'})
+bp = Blueprint('oauth2', __name__, template_folder='templates')
- def save_authorization_code(self, client_id, code, oauthreq, *args, **kwargs):
- grant = OAuth2Grant(user=oauthreq.user, client=oauthreq.client, code=code['code'],
- redirect_uri=oauthreq.redirect_uri, scopes=oauthreq.scopes)
- db.session.add(grant)
- db.session.commit()
- # Oauthlib does not really provide a way to customize grant code generation.
- # Actually `code` is created just before `save_authorization_code` is called
- # and the same dict is later used to generate the OAuth2 response. So by
- # modifing the `code` dict we can actually influence the grant code.
- code['code'] = f"{grant.id}-{code['code']}"
-
- def validate_code(self, client_id, code, client, oauthreq, *args, **kwargs):
- if '-' not in code:
- return False
- grant_id, grant_code = code.split('-', 2)
- oauthreq.grant = OAuth2Grant.query.get(grant_id)
- if not oauthreq.grant or oauthreq.grant.client != client:
- return False
- if not secrets.compare_digest(oauthreq.grant.code, grant_code):
- return False
- if oauthreq.grant.expired:
- return False
- if oauthreq.grant.user.is_deactivated:
- return False
- oauthreq.user = oauthreq.grant.user
- oauthreq.scopes = oauthreq.grant.scopes
- return True
-
- def invalidate_authorization_code(self, client_id, code, oauthreq, *args, **kwargs):
- if '-' not in code:
- return
- grant_id, grant_code = code.split('-', 2)
- grant = OAuth2Grant.query.get(grant_id)
- if not grant or grant.client != oauthreq.client:
- return
- if not secrets.compare_digest(grant.code, grant_code):
- return
- db.session.delete(grant)
- db.session.commit()
+@bp.route('/.well-known/openid-configuration')
+def discover():
+ return jsonify({
+ 'issuer': get_issuer(),
+ 'authorization_endpoint': url_for('oauth2.authorize', _external=True),
+ 'token_endpoint': url_for('oauth2.token', _external=True),
+ 'userinfo_endpoint': url_for('oauth2.userinfo', _external=True),
+ 'jwks_uri': url_for('oauth2.keys', _external=True),
+ 'scopes_supported': sorted(OIDC_SCOPES.keys()),
+ 'response_types_supported': ['code'],
+ 'grant_types_supported': ['authorization_code'],
+ 'id_token_signing_alg_values_supported': OAuth2Key.get_available_algorithms(),
+ 'subject_types_supported': ['public'],
+ 'token_endpoint_auth_methods_supported': ['client_secret_basic', 'client_secret_post'],
+ 'claims_supported': sorted(['iat', 'exp', 'aud', 'iss'] + list(OIDC_CLAIMS.keys())),
+ 'claims_parameter_supported': True,
+ 'request_uri_parameter_supported': False, # default is True
+ })
- def save_bearer_token(self, token_data, oauthreq, *args, **kwargs):
- tok = OAuth2Token(
- user=oauthreq.user,
- client=oauthreq.client,
- token_type=token_data['token_type'],
- access_token=token_data['access_token'],
- refresh_token=token_data['refresh_token'],
- expires_in_seconds=token_data['expires_in'],
- scopes=oauthreq.scopes
- )
- db.session.add(tok)
- db.session.commit()
- # Oauthlib does not really provide a way to customize access/refresh token
- # generation. Actually `token_data` is created just before
- # `save_bearer_token` is called and the same dict is later used to generate
- # the OAuth2 response. So by modifing the `token_data` dict we can actually
- # influence the tokens.
- token_data['access_token'] = f"{tok.id}-{token_data['access_token']}"
- token_data['refresh_token'] = f"{tok.id}-{token_data['refresh_token']}"
- return oauthreq.client.default_redirect_uri
-
- def validate_grant_type(self, client_id, grant_type, client, oauthreq, *args, **kwargs):
- return grant_type == 'authorization_code'
-
- def confirm_redirect_uri(self, client_id, code, redirect_uri, client, oauthreq, *args, **kwargs):
- return redirect_uri == oauthreq.grant.redirect_uri
-
- def validate_bearer_token(self, token_value, scopes, oauthreq):
- if '-' not in token_value:
- return False
- tok_id, tok_secret = token_value.split('-', 2)
- tok = OAuth2Token.query.get(tok_id)
- if not tok or not secrets.compare_digest(tok.access_token, tok_secret):
- return False
- if tok.expired:
- oauthreq.error_message = 'Token expired'
- return False
- if tok.user.is_deactivated:
- oauthreq.error_message = 'User deactivated'
- return False
- if not set(scopes).issubset(tok.scopes):
- oauthreq.error_message = 'Scopes invalid'
- return False
- oauthreq.access_token = tok
- oauthreq.user = tok.user
- oauthreq.scopes = scopes
- oauthreq.client = tok.client
- oauthreq.client_id = oauthreq.client.client_id
- return True
-
- # get_original_scopes/validate_refresh_token are only used for refreshing tokens. We don't implement the refresh endpoint.
- # revoke_token is only used for revoking access tokens. We don't implement the revoke endpoint.
- # get_id_token/validate_silent_authorization/validate_silent_login are OpenID Connect specfic.
- # validate_user/validate_user_match are not required for Authorization Code Grant flow.
-
-validator = UffdRequestValidator()
-server = oauthlib.oauth2.WebApplicationServer(validator)
-bp = Blueprint('oauth2', __name__, url_prefix='/oauth2/', template_folder='templates')
-
-@bp.errorhandler(oauthlib.oauth2.rfc6749.errors.OAuth2Error)
-def handle_oauth2error(error):
- return render_template('oauth2/error.html', error=type(error).__name__, error_description=error.description), 400
-
-@bp.route('/authorize', methods=['GET', 'POST'])
-def authorize():
- scopes, credentials = server.validate_authorization_request(request.url, request.method, request.form, request.headers)
- client = OAuth2Client.query.filter_by(client_id=credentials['client_id']).one()
+@bp.route('/oauth2/keys')
+def keys():
+ return jsonify({
+ 'keys': [key.public_key_jwks_dict for key in OAuth2Key.query.filter_by(active=True)],
+ }), 200, {'Cache-Control': ['max-age=86400, public, must-revalidate, no-transform=true']}
+
+def oauth2_redirect(**extra_args):
+ urlparts = urllib.parse.urlparse(request.oauth2_redirect_uri)
+ args = urllib.parse.parse_qs(urlparts.query)
+ if 'state' in request.args:
+ args['state'] = request.args['state']
+ for key, value in extra_args.items():
+ if value is not None:
+ args[key] = [value]
+ return redirect(urlparts._replace(query=urllib.parse.urlencode(args, doseq=True)).geturl())
+
+class OAuth2Error(Exception):
+ ERROR: str
+
+ def __init__(self, error_description=None):
+ self.error_description = error_description
+
+ @property
+ def params(self):
+ res = {'error': self.ERROR}
+ if self.error_description:
+ res['error_description'] = self.error_description
+ return res
+
+# RFC 6749: OAuth 2.0
+class InvalidRequestError(OAuth2Error):
+ ERROR = 'invalid_request'
+
+class UnsupportedResponseTypeError(OAuth2Error):
+ ERROR = 'unsupported_response_type'
+
+class InvalidScopeError(OAuth2Error):
+ ERROR = 'invalid_scope'
+
+class InvalidClientError(OAuth2Error):
+ ERROR = 'invalid_client'
+
+class UnsupportedGrantTypeError(OAuth2Error):
+ ERROR = 'unsupported_grant_type'
+
+class InvalidGrantError(OAuth2Error):
+ ERROR = 'invalid_grant'
+
+class AccessDeniedError(OAuth2Error):
+ ERROR = 'access_denied'
+
+ def __init__(self, flash_message=None, **kwargs):
+ self.flash_message = flash_message
+ super().__init__(**kwargs)
+
+# RFC 6750: OAuth 2.0 Bearer Token Usage
+class InvalidTokenError(OAuth2Error):
+ ERROR = 'invalid_token'
+
+# OpenID Connect Core 1.0
+class LoginRequiredError(OAuth2Error):
+ ERROR = 'login_required'
+
+ def __init__(self, response=None, flash_message=None, **kwargs):
+ self.response = response
+ self.flash_message = flash_message
+ super().__init__(**kwargs)
+
+class RequestNotSupportedError(OAuth2Error):
+ ERROR = 'request_not_supported'
+
+class RequestURINotSupportedError(OAuth2Error):
+ ERROR = 'request_uri_not_supported'
+
+def authorize_validate_request():
+ request.oauth2_redirect_uri = None
+ for param in request.args:
+ if len(request.args.getlist(param)) > 1:
+ raise InvalidRequestError(error_description=f'Duplicate parameter {param}')
+
+ if 'client_id' not in request.args:
+ raise InvalidRequestError(error_description='Required parameter client_id missing')
+ client_id = request.args['client_id']
+ client = OAuth2Client.query.filter_by(client_id=client_id).one_or_none()
+ if not client:
+ raise InvalidRequestError(error_description=f'Unknown client {client_id}')
+
+ redirect_uri = request.args.get('redirect_uri')
+ if redirect_uri and redirect_uri not in client.redirect_uris:
+ raise InvalidRequestError(error_description='Invalid redirect_uri')
+ request.oauth2_redirect_uri = redirect_uri or client.default_redirect_uri
+ if not request.oauth2_redirect_uri:
+ raise InvalidRequestError(error_description='Parameter redirect_uri required')
+
+ if 'response_type' not in request.args:
+ raise InvalidRequestError(error_description='Required parameter response_type missing')
+ response_type = request.args['response_type']
+ if response_type != 'code':
+ raise UnsupportedResponseTypeError(error_description='Unsupported response type')
+
+ scopes = {scope for scope in request.args.get('scope', '').split(' ') if scope} or {'profile'}
+ if scopes == {'profile'}:
+ pass # valid plain OAuth2 scopes
+ elif 'openid' in scopes:
+ # OIDC core spec: "Scope values used that are not understood by an implementation SHOULD be ignored."
+ # Since we don't support some of the optional scope values defined by the
+ # spec (phone, address, offline_access), it's probably best to ignore all
+ # unknown scopes.
+ pass # valid OIDC scopes
+ else:
+ raise InvalidScopeError(error_description='Unknown scope')
+
+ return OAuth2Grant(
+ client=client,
+ # redirect_uri is None if not present in request! This affects token request validation.
+ redirect_uri=redirect_uri,
+ scopes=scopes,
+ )
+
+def authorize_validate_request_oidc(grant):
+ nonce = request.args.get('nonce')
+ claims = json.loads(request.args['claims']) if 'claims' in request.args else None
+ if 'request' in request.args:
+ raise RequestNotSupportedError()
+ if 'request_uri' in request.args:
+ raise RequestURINotSupportedError()
+
+ prompt_values = {value for value in request.args.get('prompt', '').split(' ') if value}
+ if 'none' in prompt_values and prompt_values != {'none'}:
+ raise InvalidRequestError(error_description='Invalid usage of none prompt parameter value')
+
+ sub_value = None
+ if claims and claims.get('id_token', {}).get('sub', {}).get('value') is not None:
+ sub_value = claims['id_token']['sub']['value']
+ if 'id_token_hint' in request.args:
+ try:
+ id_token = OAuth2Key.decode_jwt(
+ request.args['id_token_hint'],
+ issuer=get_issuer(),
+ options={'verify_exp': False, 'verify_aud': False}
+ )
+ except (jwt.exceptions.InvalidTokenError, jwt.exceptions.InvalidKeyError) as err:
+ raise InvalidRequestError(error_description='Invalid id_token_hint value') from err
+ if sub_value is not None and id_token['sub'] != sub_value:
+ raise InvalidRequestError(error_description='Ambiguous sub values in claims and id_token_hint')
+ sub_value = id_token['sub']
+
+ # We "MUST only send a positive response if the End-User identified by that
+ # sub value has an active session with the Authorization Server or has been
+ # Authenticated as a result of the request". However, we currently cannot
+ # display the login page if there is already a valid session. So we can only
+ # support sub_value in combination with prompt=none for now.
+ if sub_value is not None and 'none' not in prompt_values:
+ raise InvalidRequestError(error_description='id_token_hint or sub claim value not supported without prompt=none')
+
+ grant.nonce = nonce
+ grant.claims = claims
+ return grant, sub_value, prompt_values
+
+def authorize_user(client):
if request.user:
- credentials['user'] = request.user
- elif 'devicelogin_started' in session:
+ return request.user
+
+ if 'devicelogin_started' in session:
del session['devicelogin_started']
host_delay = host_ratelimit.get_delay()
if host_delay:
- flash(_('We received too many requests from your ip address/network! Please wait at least %(delay)s.', delay=format_delay(host_delay)))
- return redirect(url_for('session.login', ref=request.full_path, devicelogin=True))
+ raise LoginRequiredError(
+ flash_message=_(
+ 'We received too many requests from your ip address/network! Please wait at least %(delay)s.',
+ delay=format_delay(host_delay)
+ ),
+ response=redirect(url_for('session.login', ref=request.full_path, devicelogin=True))
+ )
host_ratelimit.log()
initiation = OAuth2DeviceLoginInitiation(client=client)
db.session.add(initiation)
try:
db.session.commit()
- except IntegrityError:
- flash(_('Device login is currently not available. Try again later!'))
- return redirect(url_for('session.login', ref=request.values['ref'], devicelogin=True))
+ except IntegrityError as err:
+ raise LoginRequiredError(
+ flash_message=_('Device login is currently not available. Try again later!'),
+ response=redirect(url_for('session.login', ref=request.values['ref'], devicelogin=True))
+ ) from err
session['devicelogin_id'] = initiation.id
session['devicelogin_secret'] = initiation.secret
- return redirect(url_for('session.devicelogin', ref=request.full_path))
- elif 'devicelogin_id' in session and 'devicelogin_secret' in session and 'devicelogin_confirmation' in session:
- initiation = OAuth2DeviceLoginInitiation.query.filter_by(id=session['devicelogin_id'], secret=session['devicelogin_secret'],
- client=client).one_or_none()
+ raise LoginRequiredError(response=redirect(url_for('session.devicelogin', ref=request.full_path)))
+ if 'devicelogin_id' in session and 'devicelogin_secret' in session and 'devicelogin_confirmation' in session:
+ initiation = OAuth2DeviceLoginInitiation.query.filter_by(
+ id=session['devicelogin_id'],
+ secret=session['devicelogin_secret'],
+ client=client
+ ).one_or_none()
confirmation = DeviceLoginConfirmation.query.get(session['devicelogin_confirmation'])
del session['devicelogin_id']
del session['devicelogin_secret']
del session['devicelogin_confirmation']
if not initiation or initiation.expired or not confirmation or confirmation.user.is_deactivated:
- flash(_('Device login failed'))
- return redirect(url_for('session.login', ref=request.full_path, devicelogin=True))
- credentials['user'] = confirmation.user
+ raise LoginRequiredError(
+ flash_message=_('Device login failed'),
+ response=redirect(url_for('session.login', ref=request.full_path, devicelogin=True))
+ )
db.session.delete(initiation)
db.session.commit()
- else:
- flash(_('You need to login to access this service'))
- return redirect(url_for('session.login', ref=request.full_path, devicelogin=True))
-
- # Here we would normally ask the user, if he wants to give the requesting
- # service access to his data. Since we only have trusted services (the
- # clients defined in the server config), we don't ask for consent.
- if not client.access_allowed(credentials['user']):
- abort(403, description=_("You don't have the permission to access the service <b>%(service_name)s</b>.", service_name=client.service.name))
+ return confirmation.user
+
+ raise LoginRequiredError(
+ flash_message=_('You need to login to access this service'),
+ response=redirect(url_for('session.login', ref=request.full_path, devicelogin=True))
+ )
+
+@bp.route('/oauth2/authorize')
+def authorize():
+ is_oidc = 'openid' in request.args.get('scope', '').split(' ')
+
+ try:
+ grant = authorize_validate_request()
+ sub_value, prompt_values = None, []
+ if is_oidc:
+ grant, sub_value, prompt_values = authorize_validate_request_oidc(grant)
+ except OAuth2Error as err:
+ # Correct OAuth2/OIDC error handling would be to redirect back to the
+ # client with an error paramter, unless client_id or redirect_uri is
+ # invalid. However, uffd never did that before adding OIDC support and
+ # many applications fail to correctly handle this case. As a compromise
+ # we report errors correctly in OIDC mode and don't in plain OAuth2 mode.
+ if is_oidc and request.oauth2_redirect_uri:
+ return oauth2_redirect(**err.params)
+ return render_template('oauth2/error.html', **err.params), 400
+
+ try:
+ user = authorize_user(grant.client)
+ if sub_value is not None and str(user.unix_uid) != sub_value:
+ # We only reach this point in OIDC requests with prompt=none, see
+ # authorize_validate_request_oidc. So this LoginRequiredError is
+ # always returned as a redirect back to the client.
+ raise LoginRequiredError()
+ if not grant.client.access_allowed(user):
+ raise AccessDeniedError(flash_message=_(
+ "You don't have the permission to access the service <b>%(service_name)s</b>.",
+ service_name=grant.client.service.name
+ ))
+ grant.user = user
+ except LoginRequiredError as err:
+ # We abuse LoginRequiredError to signal a redirect to the login page
+ if is_oidc and 'none' in prompt_values:
+ err.error_description = 'Login required but prompt value set to none'
+ return oauth2_redirect(**err.params)
+ if err.flash_message:
+ flash(err.flash_message)
+ return err.response
+ except AccessDeniedError as err:
+ if is_oidc and request.oauth2_redirect_uri:
+ return oauth2_redirect(**err.params)
+ abort(403, description=err.flash_message)
+
session['oauth2-clients'] = session.get('oauth2-clients', [])
- if client.client_id not in session['oauth2-clients']:
- session['oauth2-clients'].append(client.client_id)
+ if grant.client.client_id not in session['oauth2-clients']:
+ session['oauth2-clients'].append(grant.client.client_id)
+ db.session.add(grant)
+ db.session.commit()
+ return oauth2_redirect(code=grant.code)
- headers, body, status = server.create_authorization_response(request.url, request.method, request.form, request.headers, scopes, credentials)
- return body or '', status, headers
+def token_authenticate_client():
+ for param in ('client_id', 'client_secret'):
+ if len(request.form.getlist(param)) > 1:
+ raise InvalidRequestError(error_description=f'Duplicate parameter {param}')
+ if request.authorization:
+ client_id = urllib.parse.unquote(request.authorization.username)
+ client_secret = urllib.parse.unquote(request.authorization.password)
+ if request.form.get('client_id', client_id) != client_id:
+ raise InvalidRequestError(error_description='Ambiguous parameter client_id')
+ if 'client_secret' in request.form:
+ raise InvalidRequestError(error_description='Ambiguous parameter client_secret')
+ elif 'client_id' in request.form and 'client_secret' in request.form:
+ client_id = request.form['client_id']
+ client_secret = request.form['client_secret']
+ else:
+ raise InvalidClientError()
+
+ client = OAuth2Client.query.filter_by(client_id=client_id).one_or_none()
+ if client is None or not client.client_secret.verify(client_secret):
+ raise InvalidClientError()
+ if client.client_secret.needs_rehash:
+ client.client_secret = client_secret
+ db.session.commit()
+ return client
+
+def token_validate_request(client):
+ for param in ('grant_type', 'code', 'redirect_uri'):
+ if len(request.form.getlist(param)) > 1:
+ raise InvalidRequestError(error_description=f'Duplicate parameter {param}')
+ if 'grant_type' not in request.form:
+ raise InvalidRequestError(error_description='Parameter grant_type missing')
+ grant_type = request.form['grant_type']
+ if grant_type != 'authorization_code':
+ raise UnsupportedGrantTypeError()
+ if 'code' not in request.form:
+ raise InvalidRequestError(error_description='Parameter code missing')
+ code = request.form['code']
-@bp.route('/token', methods=['GET', 'POST'])
+ grant = OAuth2Grant.get_by_authorization_code(code)
+ if not grant or grant.client != client:
+ raise InvalidGrantError()
+ if grant.redirect_uri and grant.redirect_uri != request.form.get('redirect_uri'):
+ raise InvalidRequestError(error_description='Parameter redirect_uri missing or invalid')
+ return grant
+
+@bp.route('/oauth2/token', methods=['POST'])
def token():
- headers, body, status = server.create_token_response(request.url, request.method, request.form,
- request.headers, {'authorization': request.authorization})
- return body, status, headers
-
-def oauth_required(*scopes):
- def wrapper(func):
- @functools.wraps(func)
- def decorator(*args, **kwargs):
- valid, oauthreq = server.verify_request(request.url, request.method, request.form, request.headers, scopes)
- if not valid:
- abort(401)
- request.oauth = oauthreq
- return func(*args, **kwargs)
- return decorator
- return wrapper
-
-@bp.route('/userinfo')
-@oauth_required('profile')
+ try:
+ client = token_authenticate_client()
+ grant = token_validate_request(client)
+ except InvalidClientError as err:
+ return jsonify(err.params), 401, {'WWW-Authenticate': ['Basic realm="oauth2"']}
+ except OAuth2Error as err:
+ return jsonify(err.params), 400
+
+ tok = grant.make_token()
+ db.session.add(tok)
+ db.session.delete(grant)
+ db.session.commit()
+
+ resp = {
+ 'token_type': 'Bearer',
+ 'access_token': tok.access_token,
+ 'expires_in': tok.EXPIRES_IN,
+ 'scope': ' '.join(tok.scopes),
+ }
+ if 'openid' in tok.scopes:
+ key = OAuth2Key.get_preferred_key()
+ id_token = render_claims(['openid'], (grant.claims or {}).get('id_token', {}), tok.service_user)
+ id_token['iss'] = get_issuer()
+ id_token['aud'] = tok.client.client_id
+ id_token['iat'] = int(time.time())
+ id_token['at_hash'] = key.oidc_hash(tok.access_token.encode('ascii'))
+ id_token['exp'] = id_token['iat'] + tok.EXPIRES_IN
+ if grant.nonce:
+ id_token['nonce'] = grant.nonce
+ resp['id_token'] = OAuth2Key.get_preferred_key().encode_jwt(id_token)
+ else:
+ # We don't support the refresh_token grant type. Due to limitations of
+ # oauthlib we always returned (disfunctional) refresh tokens in the past.
+ # We still do that for non-OIDC clients to not change behavour for
+ # existing clients.
+ resp['refresh_token'] = tok.refresh_token
+
+ return jsonify(resp), 200, {'Cache-Control': ['no-store']}
+
+def validate_access_token():
+ if len(request.headers.getlist('Authorization')) == 1 and 'access_token' not in request.values:
+ auth_type, auth_value = (request.headers['Authorization'].split(' ', 1) + [''])[:2]
+ if auth_type.lower() != 'bearer':
+ raise InvalidRequestError()
+ access_token = auth_value
+ elif len(request.values.getlist('access_token')) == 1 and 'Authorization' not in request.headers:
+ access_token = request.values['access_token']
+ else:
+ raise InvalidClientError()
+ tok = OAuth2Token.get_by_access_token(access_token)
+ if not tok:
+ raise InvalidTokenError()
+ return tok
+
+@bp.route('/oauth2/userinfo', methods=['GET', 'POST'])
def userinfo():
- service_user = ServiceUser.query.get((request.oauth.client.service_id, request.oauth.user.id))
- return jsonify(
- id=service_user.user.unix_uid,
- name=service_user.user.displayname,
- nickname=service_user.user.loginname,
- email=service_user.email,
- groups=[group.name for group in service_user.user.groups]
- )
+ try:
+ tok = validate_access_token()
+ except OAuth2Error as err:
+ # RFC 6750:
+ # If the request lacks any authentication information (e.g., the client
+ # was unaware that authentication is necessary or attempted using an
+ # unsupported authentication method), the resource server SHOULD NOT
+ # include an error code or other error information.
+ header = 'Bearer'
+ if request.headers.get('Authorization', '').lower().startswith('bearer') or 'access_token' in request.values:
+ header += f' error="{err.ERROR}"'
+ return '', 401, {'WWW-Authenticate': [header]}
+
+ service_user = tok.service_user
+ if 'openid' in tok.scopes:
+ resp = render_claims(tok.scopes, (tok.claims or {}).get('userinfo', {}), service_user)
+ else:
+ resp = {
+ 'id': service_user.user.unix_uid,
+ 'name': service_user.user.displayname,
+ 'nickname': service_user.user.loginname,
+ 'email': service_user.email,
+ 'groups': [group.name for group in service_user.user.groups],
+ }
+ return jsonify(resp), 200, {'Cache-Control': ['private']}
@bp.app_url_defaults
def inject_logout_params(endpoint, values):
@@ -255,7 +492,7 @@ def inject_logout_params(endpoint, values):
return
values['client_ids'] = ','.join(session['oauth2-clients'])
-@bp.route('/logout')
+@bp.route('/oauth2/logout')
def logout():
if not request.values.get('client_ids'):
return secure_local_redirect(request.values.get('ref', '/'))
--
GitLab