✨ Separation of voting session time and bearer token lifetime (#351)

Parent issue: sequentech/meta#762

iam/api/decorators.py

@@ -28,7 +28,7 @@ def get_auth_key(request):
             key = request.META.get('HTTP_HTTP_AUTH', None)
     return key
 
-def get_login_user(request):
+def get_login_user(request, has_expiry_timestamp = True):
     key = get_auth_key(request)
     hmac_token = None
 
@@ -39,6 +39,9 @@ def get_login_user(request):
       hmac_token = HMACToken(key)
       user = User.objects.get(username=hmac_token.get_userid())
 
+      if (hmac_token.expiry_timestamp != False) != has_expiry_timestamp:
+          return None, dict(error_codename="invalid_token_type"), hmac_token
+
       # admin auth event has a different timeout
       if user.userdata.event_id == settings.ADMIN_AUTH_ID:
           timeout = settings.ADMIN_TIMEOUT

iam/api/migrations/0054_token_duration.py

@@ -0,0 +1,23 @@
+# Generated by Felix Robles on 2024-06-27 06:36
+
+from django.db import migrations, models
+from django.contrib.postgres.fields import JSONField
+import django.core.validators
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ('api', '0053_authapi_oidc_providers'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='authevent',
+            name='refresh_token_duration_secs',
+            field=models.IntegerField(default=600, validators=[django.core.validators.MinValueValidator(0)]),
+        ),
+        migrations.AddField(
+            model_name='authevent',
+            name='access_token_duration_secs',
+            field=models.IntegerField(default=120, validators=[django.core.validators.MinValueValidator(0)]),
+        ),
+    ]

iam/api/models.py

@@ -42,7 +42,7 @@
 from django.db.models import CharField
 from django.db.models.functions import Length
 
-from utils import genhmac
+from utils import generate_access_token_hmac
 from api.middleware import CurrentUserMiddleware
 
 CharField.register_lookup(Length, 'length')
@@ -589,6 +589,30 @@ class AuthEvent(models.Model):
     )
     based_in = models.IntegerField(null=True) # auth_event_id
 
+
+    refresh_token_duration_secs = models.IntegerField(
+        default=0,
+        validators=[
+            MinValueValidator(0)
+        ]
+    )
+
+    access_token_duration_secs = models.IntegerField(
+        default=0,
+        validators=[
+            MinValueValidator(0)
+        ]
+    )
+
+    def get_refresh_token_duration_secs(self):
+        if self.refresh_token_duration_secs > 0:
+            return self.refresh_token_duration_secs 
+        is_admin = settings.ADMIN_AUTH_ID == self.id
+        return settings.ADMIN_TIMEOUT if is_admin else settings.REFRESH_TIMEOUT
+
+    def get_access_token_duration_secs(self):
+        return self.access_token_duration_secs if self.access_token_duration_secs > 0 else settings.ACCESS_TIMEOUT
+
     # will return true if allow_user_resend is defined and it's True,
     # false otherwise
     def check_allow_user_resend(self):
@@ -1224,9 +1248,9 @@ def serialize(self):
         }
         return d
 
-    def get_hmac(self):
+    def get_hmac(self, validity):
         msg = ':'.join((self.user.user.username, self.object_type, str(self.object_id), self.perm))
-        khmac = genhmac(settings.SHARED_SECRET, msg)
+        khmac = generate_access_token_hmac(settings.SHARED_SECRET, msg, validity)
         return khmac
 
     def __str__(self):

iam/api/tasks.py

@@ -25,7 +25,7 @@
 
 import plugins
 from authmethods.sms_provider import SMSProvider
-from utils import send_codes, genhmac, reproducible_json_dumps
+from utils import send_codes, generate_access_token_hmac, reproducible_json_dumps
 from .models import Action, AuthEvent, BallotBox, TallySheet
 
 logger = get_task_logger(__name__)
@@ -156,9 +156,10 @@ def launch_tally(auth_event):
         callback_url,
         json=[],
         headers={
-            'Authorization': genhmac(
+            'Authorization': generate_access_token_hmac(
                 settings.SHARED_SECRET,
-                "1:AuthEvent:%s:tally" % auth_event.id
+                "1:AuthEvent:%s:tally" % auth_event.id,
+                auth_event.get_refresh_token_duration_secs()
             ),
             'Content-type': 'application/json'
         }
@@ -244,9 +245,10 @@ def launch_virtual_tally(auth_event):
         callback_url,
         json=reproducible_json_dumps({}),
         headers={
-            'Authorization': genhmac(
+            'Authorization': generate_access_token_hmac(
                 settings.SHARED_SECRET,
-                "1:AuthEvent:%s:tally" % auth_event.id
+                "1:AuthEvent:%s:tally" % auth_event.id,
+                auth_event.get_refresh_token_duration_secs()
             ),
             'Content-type': 'application/json'
         }
@@ -545,9 +547,10 @@ def update_ballot_boxes_config(auth_event_id):
             callback_url,
             json=ballot_boxes_config,
             headers={
-                'Authorization': genhmac(
+                'Authorization': generate_access_token_hmac(
                     settings.SHARED_SECRET,
-                    "1:AuthEvent:%s:update-ballot-boxes-results-config" % auth_event_id
+                    "1:AuthEvent:%s:update-ballot-boxes-results-config" % auth_event_id,
+                    auth_event_id.get_refresh_token_duration_secs()
                 ),
                 'Content-type': 'application/json'
             }
@@ -618,9 +621,10 @@ def calculate_results_task(user_id, event_id_list):
             callback_url,
             data=config,
             headers={
-                'Authorization': genhmac(
+                'Authorization': generate_access_token_hmac(
                     settings.SHARED_SECRET,
-                    "1:AuthEvent:%s:calculate-results" % auth_event_id
+                    "1:AuthEvent:%s:calculate-results" % auth_event_id,
+                    auth_event.get_refresh_token_duration_secs()
                 ),
                 'Content-type': 'application/json'
             }
@@ -732,9 +736,10 @@ def publish_results_task(user_id, auth_event_id, visit_children, parent_auth_eve
             callback_url,
             json=data,
             headers={
-                'Authorization': genhmac(
+                'Authorization': generate_access_token_hmac(
                     settings.SHARED_SECRET,
-                    "1:AuthEvent:%s:publish-results" % auth_event_id
+                    "1:AuthEvent:%s:publish-results" % auth_event_id,
+                    auth_event.get_refresh_token_duration_secs()
                 ),
                 'Content-type': 'application/json'
             }
@@ -840,9 +845,10 @@ def unpublish_results_task(user_id, auth_event_id, parent_auth_event_id=None):
             callback_url,
             json=data,
             headers={
-                'Authorization': genhmac(
+                'Authorization': generate_access_token_hmac(
                     settings.SHARED_SECRET,
-                    "1:AuthEvent:%s:publish-results" % auth_event_id
+                    "1:AuthEvent:%s:publish-results" % auth_event_id,
+                    auth_event.get_refresh_token_duration_secs()
                 ),
                 'Content-type': 'application/json'
             }
@@ -955,9 +961,10 @@ def set_public_candidates_task(
             callback_url,
             json=data,
             headers={
-                'Authorization': genhmac(
+                'Authorization': generate_access_token_hmac(
                     settings.SHARED_SECRET,
-                    "1:AuthEvent:%s:set-public-candidates" % auth_event_id
+                    "1:AuthEvent:%s:set-public-candidates" % auth_event_id,
+                    auth_event.get_refresh_token_duration_secs()
                 ),
                 'Content-type': 'application/json'
             }
@@ -1094,9 +1101,10 @@ def get_parent_event(parent_event_id, auth_event):
                 callback_url,
                 json=data,
                 headers={
-                    'Authorization': genhmac(
+                    'Authorization': generate_access_token_hmac(
                         settings.SHARED_SECRET,
-                        f"1:AuthEvent:{current_event_id}:{action_name}"
+                        f"1:AuthEvent:{current_event_id}:{action_name}",
+                        current_event.get_refresh_token_duration_secs()
                     ),
                     'Content-type': 'application/json'
                 }

iam/api/test_data.py

@@ -1554,7 +1554,7 @@ def generate_auth_event_and_voter_credentials(auth_method):
     authentication method.
     '''
     from api.models import AuthEvent
-    from authmethods.utils import genhmac
+    from authmethods.utils import generate_access_token_hmac
     from authmethods.models import Code
     auth_event = None
     voter = None
@@ -1834,14 +1834,15 @@ def generate_auth_event_and_voter_credentials(auth_method):
         voter.userdata.save()
 
         credentials = {
-          'auth-token': genhmac(
+          'auth-token': generate_access_token_hmac(
             key=settings.SHARED_SECRET,
             msg=':'.join([
                 voter.userdata.metadata['user_id'],
                 'AuthEvent',
                 str(auth_event.id),
                 'vote'
-            ])
+            ]),
+            validity=auth_event.refresh_token_duration_secs
           )
         }
     else:

iam/api/tests.py

@@ -3101,14 +3101,16 @@ def setUp(self):
         self.u = u.userdata
         self.uid = u.id
 
-    def genhmac(self, key, msg):
+    def generate_access_token_hmac(self, key, msg):
         import hmac
 
         if not key or not msg:
            return
 
         timestamp = int(timezone.now().timestamp())
-        msg = "%s:%s" % (msg, str(timestamp))
+        validity = 120
+        expiry_timestamp = timestamp + validity
+        msg = "%s:%s:%s:%s" % (msg, str(expiry_timestamp), "timeout-token", str(timestamp))
 
         h = hmac.new(key, msg.encode('utf-8'), "sha256")
         return 'khmac:///sha-256;' + h.hexdigest() + '/' + msg
@@ -3117,7 +3119,7 @@ def genhmac(self, key, msg):
     def test_callback(self):
         c = JClient()
         timed_auth = "test:AuthEvent:%d:Callback" % self.aeid
-        hmac = self.genhmac(settings.SHARED_SECRET, timed_auth)
+        hmac = self.generate_access_token_hmac(settings.SHARED_SECRET, timed_auth)
         c.set_auth_token(hmac)
         response = c.post('/api/auth-event/%d/callback/' % self.aeid, {})
         self.assertEqual(response.status_code, 200)
@@ -3275,11 +3277,13 @@ class TestRevotes(TestCase):
     def setUpTestData():
         flush_db_load_fixture()
 
-    def genhmac(self, key, msg):
+    def generate_access_token_hmac(self, key, msg):
         import hmac
         import datetime
         timestamp = int(timezone.now().timestamp())
-        msg = "%s:%s" % (msg, str(timestamp))
+        validity = 120
+        expiry_timestamp = timestamp + validity
+        msg = "%s:%s:%s:%s" % (msg, str(expiry_timestamp), "timeout-token", str(timestamp))
 
         h = hmac.new(key, msg.encode('utf-8'), "sha256")
         return 'khmac:///sha-256;' + h.hexdigest() + '/' + msg
@@ -3348,7 +3352,7 @@ def test_check_1_2_revotes(self):
         self.assertEqual(response.status_code, 200)
         response = c.authenticate(self.aeid, test_data.auth_email_default1)
         self.assertEqual(response.status_code, 200)
-        auth_token = self.genhmac(settings.SHARED_SECRET, "%s:AuthEvent:%d:RegisterSuccessfulLogin" % (cuser.username, self.aeid))
+        auth_token = self.generate_access_token_hmac(settings.SHARED_SECRET, "%s:AuthEvent:%d:RegisterSuccessfulLogin" % (cuser.username, self.aeid))
         c.set_auth_token(auth_token)
         response = c.post('/api/auth-event/%d/successful_login/%s' % (self.aeid, cuser.username), {})
         self.assertEqual(response.status_code, 200)
@@ -3364,7 +3368,7 @@ def test_check_1_2_revotes(self):
         self.assertEqual(response.status_code, 200)
         response = c.authenticate(self.aeid, test_data.auth_email_default1)
         self.assertEqual(response.status_code, 200)
-        auth_token = self.genhmac(settings.SHARED_SECRET, "%s:AuthEvent:%d:RegisterSuccessfulLogin" % (cuser.username, self.aeid))
+        auth_token = self.generate_access_token_hmac(settings.SHARED_SECRET, "%s:AuthEvent:%d:RegisterSuccessfulLogin" % (cuser.username, self.aeid))
         c.set_auth_token(auth_token)
         response = c.post('/api/auth-event/%d/successful_login/%s' % (self.aeid, cuser.username), {})
         self.assertEqual(response.status_code, 200)
@@ -3394,7 +3398,7 @@ def test_check_50_revotes_max(self):
         for i in range(0, 50):
             response = c.authenticate(self.aeid, test_data.auth_email_default1)
             self.assertEqual(response.status_code, 200)
-            auth_token = self.genhmac(settings.SHARED_SECRET, "%s:AuthEvent:%d:RegisterSuccessfulLogin" % (cuser.username, self.aeid))
+            auth_token = self.generate_access_token_hmac(settings.SHARED_SECRET, "%s:AuthEvent:%d:RegisterSuccessfulLogin" % (cuser.username, self.aeid))
             c.set_auth_token(auth_token)
             response = c.post('/api/auth-event/%d/successful_login/%s' % (self.aeid, cuser.username), {})
 

iam/api/views.py

@@ -57,7 +57,7 @@
     update_alt_methods_config,
     check_alt_auth_methods,
     check_admin_fields,
-    genhmac,
+    generate_access_token_hmac,
     HMACToken,
     json_response,
     paginate,
@@ -670,7 +670,6 @@ def post(self, request, pk):
             )
 authenticate = Authenticate.as_view()
 
-
 class AuthenticateOtl(View):
     ''' Authenticate into the iam '''
 
@@ -791,12 +790,12 @@ def get(self, request, pk):
         data = {}
 
         if u and error is None:
-            data = {
-              'auth-token': genhmac(settings.SHARED_SECRET, u.username)
-            }
+            data = {}
             auth_event = get_object_or_404(AuthEvent, pk=pk)
             req = {}
             auth_data = return_auth_data('Ping', req, request, u, auth_event)
+            if 'auth-token' in auth_data:
+                data['auth-token'] = auth_data['auth-token']
             if 'vote-permission-token' in auth_data:
                 data['vote-permission-token'] = auth_data['vote-permission-token']
             if 'vote-children-info' in auth_data:
@@ -1236,8 +1235,9 @@ def post(self, request):
                 error_codename=ErrorCodes.BAD_REQUEST)
 
         msg = ':'.join((request.user.username, object_type, str(obj_id), filtered_perms))
+        auth_event = request.user.userdata.event
 
-        data['permission-token'] = genhmac(settings.SHARED_SECRET, msg)
+        data['permission-token'] = generate_access_token_hmac(settings.SHARED_SECRET, msg, auth_event.get_access_token_duration_secs())
         return json_response(data)
 getperms = login_required(GetPerms.as_view())
 
@@ -1908,7 +1908,7 @@ def post(request, pk=None):
             action.save()
 
 
-        data = {'status': 'ok', 'id': ae.pk, 'perm': acl.get_hmac()}
+        data = {'status': 'ok', 'id': ae.pk, 'perm': acl.get_hmac(ae.get_refresh_token_duration_secs())}
         return json_response(data)
 
     def get(self, request, pk=None):

iam/authmethods/m_dnie.py

@@ -92,7 +92,7 @@ def dnie_auth(request, authid):
     #    u = ud.user
 
     #msg = ':'.join((u.username, "AuthEvent", ae.id, "vote"))
-    #khmac = genhmac(settings.SHARED_SECRET, msg)
+    #khmac = generate_access_token_hmac(settings.SHARED_SECRET, msg, ae.refresh_token_duration_secs)
     #head, path = khmac.split(";")[1]
     #array = path.split("/")
     #hash, msg = array[0], array[1]

iam/authmethods/m_smart_link.py

@@ -372,7 +372,7 @@ def authenticate(self, auth_event, request):
             error_codename=SmartLinkErrorCodes.INVALID_PERMISSION
         )
 
-      if not hmac_token.check_expiration(settings.TIMEOUT):
+      if not hmac_token.check_expiration(settings.SMARTLINK_TIMEOUT):
         return self.error(
             SmartLinkErrorCodes.EXPIRED_AUTH_TOKEN,
             auth_event=auth_event,
@@ -395,7 +395,7 @@ def authenticate(self, auth_event, request):
       verified = verifyhmac(
         key=shared_secret,
         msg=hmac_token.msg,
-        seconds=settings.TIMEOUT,
+        seconds=settings.SMARTLINK_TIMEOUT,
         at=hmac_token
       )