Add groups to users automatically via Oauth #1677

__init__.py

@@ -0,0 +1,17 @@
+from pyramid.view import view_config
+from pyramid.security import remember, forget
+from pyramid.httpexceptions import HTTPFound, HTTPUnauthorized
+
+@view_config(route_name='login', request_method='POST')
+def login(request):
+    username = request.params.get('username')
+    password = request.params.get('password')
+
+    if username is None or password is None:
+        return HTTPUnauthorized('Username and password are required')
+
+    if check_credentials(username, password):
+        headers = remember(request, username)
+        return HTTPFound(location="/", headers=headers)
+
+    return HTTPUnauthorized('Invalid username or password')

package.json

@@ -148,6 +148,7 @@
   },
   "dependencies": {
     "btoa": "^1.2.1",
+    "kinto": "^15.0.0",
     "uuid": "^9.0.0"
   },
   "devDependencies": {

src/service/authentication.py

@@ -0,0 +1,40 @@
+import pytest
+from pyramid import testing
+from myproject.authentication import GroupAwareAuthenticationPolicy
+from jwt import ExpiredSignatureError, DecodeError
+import jwt
+
+@pytest.fixture
+def request_with_auth_header():
+  request = testing.DummyRequest()
+  request.headers['Authorization'] = 'Bearer validtoken123'
+  return request
+
+@pytest.fixture
+def policy():
+  return GroupAwareAuthenticationPolicy('secretkey')
+
+def test_valid_token(request_with_auth_header, policy):
+  with pytest.mock.patch('jwt.decode') as mock_decode:
+    mock_decode.return_value = {'sub': 'user123', 'groups': ['group1', 'group2']}
+    userid = policy.unauthenticated_userid(request_with_auth_header)
+    assert userid == 'user123'
+    principals = policy.effective_principals(request_with_auth_header)
+    assert 'group:group1' in principals
+    assert 'group:group2' in principals
+
+def test_expired_token(policy):
+  request = testing.DummyRequest()
+  request.headers['Authorization'] = 'Bearer expiredtoken123'
+  with pytest.mock.patch('jwt.decode', side_effect=ExpiredSignatureError):
+    userid = policy.unauthenticated_userid(request)
+    assert userid is None
+
+def test_malformed_token(policy):
+  request = testing.DummyRequest()
+  request.headers['Authorization'] = 'Bearer notatoken'
+  with pytest.mock.patch('jwt.decode', side_effect=DecodeError):
+    userid = policy.unauthenticated_userid(request)
+    assert userid is None
+    assert not policy.effective_principals(request)  # Only default principal should be present
+

test/authentication_tests.py

@@ -0,0 +1,37 @@
+import unittest
+from pyramid import testing
+from myproject.authentication import GroupAwareAuthenticationPolicy
+import jwt
+
+class AuthenticationPolicyTests(unittest.TestCase):
+    def setUp(self):
+        self.config = testing.setUp()
+        self.policy = GroupAwareAuthenticationPolicy('secretkey')
+
+    def tearDown(self):
+        testing.tearDown()
+
+    def test_valid_token(self):
+        request = testing.DummyRequest()
+        request.headers['Authorization'] = 'Bearer validtoken123'
+
+        with unittest.mock.patch('jwt.decode') as mock_decode:
+            mock_decode.return_value = {'sub': 'user123', 'groups': ['group1', 'group2']}
+            userid = self.policy.unauthenticated_userid(request)
+            self.assertEqual(userid, 'user123')
+            principals = self.policy.effective_principals(request)
+            self.assertIn('group:group1', principals)
+            self.assertIn('group:group2', principals)
+
+    def test_expired_token(self):
+        request = testing.DummyRequest()
+        request.headers['Authorization'] = 'Bearer expiredtoken123'
+
+        with unittest.mock.patch('jwt.decode', side_effect=jwt.ExpiredSignatureError):
+            userid = self.policy.unauthenticated_userid(request)
+            self.assertIsNone(userid)
+
+# More tests can be added here
+
+if __name__ == '__main__':
+    unittest.main()