tests

config/database_versions.py

@@ -1,5 +1,6 @@
 # Used for automatic database upgrades and version checks
 DATABASE_VERSIONS = [
+    '8.10.0',
     '8.9.0',
     '8.8.0',
     '8.6.0',

install/1_structure.sql

@@ -17,6 +17,7 @@ SET client_min_messages = warning;
 SET row_security = off;
 
 ALTER TABLE IF EXISTS ONLY web.user_tokens DROP CONSTRAINT IF EXISTS user_tokens_user_id_fkey;
+ALTER TABLE IF EXISTS ONLY web.user_tokens DROP CONSTRAINT IF EXISTS user_tokens_creator_id_fkey;
 ALTER TABLE IF EXISTS ONLY web.user_settings DROP CONSTRAINT IF EXISTS user_settings_user_id_fkey;
 ALTER TABLE IF EXISTS ONLY web.user_notes DROP CONSTRAINT IF EXISTS user_notes_user_id_fkey;
 ALTER TABLE IF EXISTS ONLY web.user_notes DROP CONSTRAINT IF EXISTS user_notes_entity_id_fkey;
@@ -1545,13 +1546,14 @@ ALTER SEQUENCE web.user_settings_id_seq OWNED BY web.user_settings.id;
 CREATE TABLE web.user_tokens (
     id integer NOT NULL,
     user_id integer NOT NULL,
+    creator_id integer NOT NULL,
     name text,
     jti text,
     valid_from timestamp without time zone,
     valid_until timestamp without time zone,
     created timestamp without time zone DEFAULT now() NOT NULL,
     modified timestamp without time zone,
-    revoked boolean NOT NULL DEFAULT false
+    revoked boolean DEFAULT false NOT NULL
 );
 
 
@@ -2699,12 +2701,20 @@ ALTER TABLE ONLY web.user_settings
     ADD CONSTRAINT user_settings_user_id_fkey FOREIGN KEY (user_id) REFERENCES web."user"(id) ON UPDATE CASCADE ON DELETE CASCADE;
 
 
+--
+-- Name: user_tokens user_tokens_creator_id_fkey; Type: FK CONSTRAINT; Schema: web; Owner: openatlas
+--
+
+ALTER TABLE ONLY web.user_tokens
+    ADD CONSTRAINT user_tokens_creator_id_fkey FOREIGN KEY (creator_id) REFERENCES web."user"(id) ON UPDATE CASCADE;
+
+
 --
 -- Name: user_tokens user_tokens_user_id_fkey; Type: FK CONSTRAINT; Schema: web; Owner: openatlas
 --
 
 ALTER TABLE ONLY web.user_tokens
-    ADD CONSTRAINT user_tokens_user_id_fkey FOREIGN KEY (user_id) REFERENCES web."user"(id) ON UPDATE CASCADE ON DELETE SET NULL;
+    ADD CONSTRAINT user_tokens_user_id_fkey FOREIGN KEY (user_id) REFERENCES web."user"(id) ON UPDATE CASCADE;
 
 
 --

install/upgrade/8.10.0.sql

@@ -19,8 +19,8 @@ CREATE TABLE IF NOT EXISTS web.user_tokens
     modified timestamp without time zone,
     revoked boolean NOT NULL DEFAULT false,
     CONSTRAINT user_tokens_pkey PRIMARY KEY (id),
-    CONSTRAINT user_tokens_user_id_fkey FOREIGN KEY (user_id) REFERENCES web."user" (id) MATCH SIMPLE ON UPDATE CASCADE ON DELETE CASCADE,
-    CONSTRAINT user_tokens_creator_id_fkey FOREIGN KEY (creator_id) REFERENCES web."user" (id) MATCH SIMPLE ON UPDATE CASCADE ON DELETE CASCADE
+    CONSTRAINT user_tokens_user_id_fkey FOREIGN KEY (user_id) REFERENCES web."user" (id) MATCH SIMPLE ON UPDATE CASCADE ON DELETE NO ACTION,
+    CONSTRAINT user_tokens_creator_id_fkey FOREIGN KEY (creator_id) REFERENCES web."user" (id) MATCH SIMPLE ON UPDATE CASCADE ON DELETE NO ACTION
 );
 
 ALTER TABLE IF EXISTS web.user_tokens OWNER to openatlas;

openatlas/__init__.py

@@ -134,11 +134,12 @@ def teardown_request(_exception: Optional[Any]) -> None:
 
 
 @jwt.token_in_blocklist_loader
-def check_if_token_revoked(
+def check_incoming_tokens(
         jwt_header: dict[str, Any],
         jwt_payload: dict[str, Any]) -> bool:
     if not jwt_header['typ'] == 'JWT':
         return True
+    # todo: add user active check. maybe add it to sql query
     token_ = check_token_revoked(jwt_payload["jti"])
     if token_['revoked'] or token_['valid_until'] < datetime.datetime.now():
         return True

openatlas/views/token.py

@@ -91,7 +91,7 @@ def api_token(user_id: int = 0) -> str | Response:
     tabs['token'].buttons.append(
         button(
             _('delete revoked tokens'),
-            url_for('delete_all_tokens'),
+            url_for('delete_revoked_tokens'),
             onclick=f"return confirm('{_('delete all revoked tokens')}?')"))
     tabs['token'].buttons.append(
         button(
@@ -212,7 +212,7 @@ def delete_token(id_: int) -> str | Response:
 
 @app.route('/admin/api_token/delete_revoked_tokens/')
 @login_required
-def delete_all_tokens() -> str | Response:
+def delete_revoked_tokens() -> str | Response:
     Token.delete_all_revoked_tokens()
     flash(_('all revoked tokens deleted'), 'info')
     return redirect(f"{url_for('api_token')}")
@@ -238,5 +238,5 @@ def revoke_all_tokens() -> str | Response:
 @login_required
 def authorize_all_tokens() -> str | Response:
     Token.authorize_all_tokens()
-    flash(_('all tokens revoked'), 'info')
+    flash(_('all tokens authorize'), 'info')
     return redirect(f"{url_for('api_token')}")

tests/base.py

@@ -85,14 +85,6 @@ def get_class_mapping(data: dict[str, Any], locale: str) -> bool:
                 and data['results'][0]['label'])
 
 
-class ProfileTestCase(TestBaseCase):
-    def setUp(self) -> None:
-        super().setUp()
-        with app.test_request_context():
-            app.preprocess_request()
-            self.token_id = get_tokens(self.alice_id)
-
-
 class ExportImportTestCase(TestBaseCase):
     def setUp(self) -> None:
         super().setUp()

tests/test_profile.py

@@ -1,7 +1,7 @@
 from flask import url_for
 from flask_jwt_extended import decode_token
 
-from openatlas import app, check_if_token_revoked
+from openatlas import app, check_incoming_tokens
 from openatlas.database.token import get_tokens
 from tests.base import ProfileTestCase
 
@@ -66,78 +66,3 @@ def test_profile(self) -> None:
             data=data,
             follow_redirects=True)
         assert b'too short' in rv.data
-
-        expire_dates = [
-            {'expiration': 0, 'token_name': 'one day token'},
-            {'expiration': 1, 'token_name': '90 days token'},
-            {'expiration': 2, 'token_name': 'indefinite token'}]
-        for expire_date in expire_dates:
-            rv = c.post(
-                url_for('generate_token'),
-                data=expire_date,
-                follow_redirects=True)
-            assert b'Token stored' in rv.data
-
-        # This should get the token table, which will not be generated
-        rv = c.get(url_for('profile_index'))
-        assert b'Generate' in rv.data
-
-        # Getting valid JWT token
-        jwt_token = ''
-        rv = c.post(
-            url_for('generate_token'),
-            data={'expiration': 0, 'token_name': 'one day token'})
-        for part in rv.headers['Set-Cookie'].split(';'):
-            if 'jwt_token=' in part:
-                jwt_token = part.replace('jwt_token=', '').strip()
-                break
-
-        with app.test_request_context():
-            app.preprocess_request()
-            decoded = decode_token(jwt_token)
-            check = check_if_token_revoked({'typ': 'JWT'}, decoded)
-            assert check is False
-
-            check = check_if_token_revoked({'typ': 'Unknown'}, decoded)
-            assert check is True
-
-            tokens = get_tokens(self.alice_id)
-            token_id = tokens[0]['id']
-            token_to_be_revoked_id = 0
-            for token in tokens:
-                if token['jti'] == decoded['jti']:
-                    token_to_be_revoked_id = token['id']
-                    break
-
-        rv = c.get(
-            url_for('api_04.class_mapping', locale='de'),
-            headers={'Authorization': f'Bearer {jwt_token}'})
-        assert b'results' in rv.data
-
-        rv = c.get(
-            url_for('revoke_token', id_=token_to_be_revoked_id),
-            follow_redirects=True)
-        assert b'Token revoked' in rv.data
-
-        with app.test_request_context():
-            app.preprocess_request()
-            check = check_if_token_revoked({'typ': 'JWT'}, decoded)
-            assert check is True
-
-        rv = c.get(
-            url_for('authorize_token', id_=token_id),
-            follow_redirects=True)
-        assert b'Token authorized' in rv.data
-
-        rv = c.get(url_for('revoke_all_tokens'), follow_redirects=True)
-        assert b'All tokens revoked' in rv.data
-
-        rv = c.get(
-            url_for('delete_all_tokens'),
-            follow_redirects=True)
-        assert b'All revoked tokens deleted' in rv.data
-
-        with app.test_request_context():
-            app.preprocess_request()
-            check = check_if_token_revoked({'typ': 'JWT'}, decoded)
-            assert check is True

tests/test_token.py

@@ -1,26 +1,105 @@
-from pathlib import Path
+from flask import url_for
+from flask_jwt_extended import decode_token
 
-from flask import g, url_for
-
-from openatlas import app
-from openatlas.database import entity as db
-from openatlas.forms.util import form_to_datetime64
-from openatlas.models.entity import Link
-from tests.base import TestBaseCase, get_hierarchy, insert
+from database.token import get_tokens
+from models.user import User
+from openatlas import app, check_incoming_tokens
+from tests.base import TestBaseCase
 
 
 class TokenTests(TestBaseCase):
     def test_token(self) -> None:
         c = self.client
         with app.test_request_context():
             app.preprocess_request()
-
+            print([(user.username, user.id) for user in User.get_all()])
 
         rv = c.get(url_for('api_token'))
         assert b'Token' in rv.data
 
         rv = c.get(url_for('generate_token'))
-        assert b'Generate token' in rv.data
+        assert b'generate token' in rv.data
+
+        generating_tokens = [
+            {'expiration': 1, 'token_name': 'one day token', 'user': 2},
+            {'expiration': 20, 'token_name': '20 day token', 'user': 5},
+            {'expiration': 90, 'token_name': '90 day token', 'user': 4},
+            {'expiration': 30, 'token_name': '30 day token', 'user': 3},
+            {'expiration': 0, 'token_name': 'indefinite token', 'user': 1}]
+
+        jwt_token_strings = []
+
+        for token in generating_tokens:
+            rv = c.post(
+                url_for('generate_token'),
+                data=token)
+            for part in rv.headers['Set-Cookie'].split(';'):
+                if 'jwt_token=' in part:
+                    jwt_token_strings.append(
+                        part.replace('jwt_token=', '').strip())
+                    break
+
+        for token in generating_tokens:
+            rv = c.post(
+                url_for('generate_token'),
+                data=token,
+                follow_redirects=True)
+            assert b'Token stored' in rv.data
+
+        rv = c.post(
+            url_for('api_token'),
+            data={'user_id': 1, 'revoked': 'all', 'valid': 'all'})
+        assert b'indefinite token' in rv.data
+
+        with app.test_request_context():
+            app.preprocess_request()
+            tokens = get_tokens(0, 'all', 'all')
+
+        rv = c.get(
+            url_for('revoke_token', id_=tokens[0]['id']),
+            follow_redirects=True)
+        assert b'Token revoked' in rv.data
+
+        rv = c.get(
+            url_for('authorize_token', id_=tokens[0]['id']),
+            follow_redirects=True)
+        assert b'Token authorized' in rv.data
+
+        rv = c.get(url_for('revoke_all_tokens'), follow_redirects=True)
+        assert b'All tokens revoked' in rv.data
+
+        rv = c.get(url_for('authorize_all_tokens'), follow_redirects=True)
+        assert b'All tokens authorize' in rv.data
+
+        rv = c.get(
+            url_for('revoke_token', id_=tokens[0]['id']),
+            follow_redirects=True)
+        assert b'Token revoked' in rv.data
+
+        with app.test_request_context():
+            app.preprocess_request()
+
+            for token in jwt_token_strings:
+                decoded = decode_token(token)
+                #check = check_incoming_tokens({'typ': 'JWT'}, decoded)
+                #assert check is True if decoded['sub'] in ['Alice', 'Inactive'] else False
+
+                # should fail if check_incoming_tokens checks for user active
+                rv = c.get(
+                    url_for('api_04.class_mapping', locale='de'),
+                    headers={'Authorization': f'Bearer {token}'})
+                assert b'results' in rv.data
+
+            assert check_incoming_tokens({'typ': 'Unknown'}, decoded) is True
+
+        rv = c.get(url_for('delete_revoked_tokens'), follow_redirects=True)
+        assert b'All revoked tokens deleted' in rv.data
+
+        rv = c.get(url_for('delete_invalid_tokens'), follow_redirects=True)
+        assert b'All invalid tokens deleted' in rv.data
+
+        rv = c.get(
+            url_for('delete_token', id_=tokens[-1]['id']),
+            follow_redirects=True)
+        assert b'Token deleted' in rv.data
 
-        rv = c.post(url_for('generate_token'))
-        assert b'Generate token' in rv.data