diff --git a/README.md b/README.md
index f1cd4f309c75460fd5cb38f7bc7943caa70685e5..2c51d402193a9a58c7dcf7f5f9c51b156088bf4a 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 0e498d6a15f3d2fe548a96bc6204fcc630f7fc79..a9a87dd55c0dc858213f318844900b24cbe18172 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 794373b3a024cd8586babf297ac09b7651dacb1f..189b301027b6a4d242ad3e766b07d7160f5cfd0f 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 7200267b7ba403c61850cf91d1ff2fe175be898b..226e75ee6d31dcbeba220975937355c26ddc7f34 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 0000000000000000000000000000000000000000..52b66299c2d2e550e70e762e33cc02653b3bed5a
--- /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 42067d583ce6fd870e85a729ea877166c4745f58..2451af51519eb23b17ac0d2ab8225523ad7ae0d9 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 0000000000000000000000000000000000000000..c7a97b2f0ba1659224e5b63a7a8785a86bbde89f
--- /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 5fe24173b3ce2f5782e9fa089c5d21b50533833a..756a97212bc4b7d819700ddd1f313a481b81a1e3 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 b79dd79286a17ce34a239e01b3c9701a78331243..e8fd0cdd0a79cf5669d13c9fcf6f3d2c065c228a 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 1bed5cfc4aa37a8ab473e739d3a79d68027c4760..c37f6b37322d482fd1a09f914924024aeb661f01 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 becb485a6409de77dfd23ba606fc378f43eb138c..f46305a12ba92eec0abd2a6b27d21ad34395ab1a 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', '/'))