ERP5Security: Add optionnal expiration time on JSON Web Token

product/ERP5Security/ERP5JSONWebTokenPlugin.py

@@ -26,7 +26,7 @@
 #
 ##############################################################################
 
-from datetime import datetime
+from datetime import datetime, timedelta
 from urlparse import urlparse
 from os import urandom
 from zLOG import LOG, INFO, ERROR
@@ -80,6 +80,8 @@ class ERP5JSONWebTokenPlugin(ERP5UserManager):
   manage_options = ( ( { 'label': 'Update Secret',
                           'action': 'manage_updateERP5JSONWebTokenPluginForm', }
                         ,
+                        { 'label': 'Set Expiration Time',
+                          'action': 'manage_setERP5JSONWebTokenPluginExtpirationDelayForm', }
                       )
                       + BasePlugin.manage_options
                     )
@@ -88,6 +90,7 @@ class ERP5JSONWebTokenPlugin(ERP5UserManager):
   def __init__(self, *args, **kw):
     super(ERP5JSONWebTokenPlugin, self).__init__(*args, **kw)
     self.manage_updateERP5JSONWebTokenPlugin()
+    self.manage_setERP5JSONWebTokenPluginExtpirationDelay(0)
 
   ####################################
   #ILoginPasswordHostExtractionPlugin#
@@ -189,6 +192,9 @@ class ERP5JSONWebTokenPlugin(ERP5UserManager):
         "http_only": True,
       }
 
+      if self.expiration_delay:
+        data["exp"] = datetime.utcnow() + timedelta(seconds=self.expiration_delay)
+
       request = self.REQUEST
 
       new_cors_origin = request.form.get('new_cors_origin')
@@ -251,4 +257,26 @@ class ERP5JSONWebTokenPlugin(ERP5UserManager):
                         % (self.absolute_url(), message)
                         )
 
+  manage_setERP5JSONWebTokenPluginExtpirationDelayForm = PageTemplateFile(
+      'www/ERP5Security_setERP5JSONWebTokenPluginExtpirationDelay',
+      globals(),
+      __name__='manage_setERP5JSONWebTokenPluginExtpirationDelayForm')
+
+  security.declareProtected(ManageUsers, 'manage_setERP5JSONWebTokenPluginExtpirationDelay')
+  def manage_setERP5JSONWebTokenPluginExtpirationDelay(
+      self,
+      expiration_delay,
+      RESPONSE=None):
+    """Edit the object"""
+
+    self.expiration_delay = int(float(expiration_delay))
+
+    #Redirect
+    if RESPONSE is not None:
+      message = "Expiration Delay Set"
+      RESPONSE.redirect('%s/manage_setERP5JSONWebTokenPluginExtpirationDelayForm'
+                        '?manage_tabs_message=%s'
+                        % (self.absolute_url(), message)
+                        )
+
 InitializeClass(ERP5JSONWebTokenPlugin)

product/ERP5Security/tests/testERP5JSONWebTokenPlugin.py

@@ -28,10 +28,12 @@
 ##############################################################################
 
 import base64
+import jwt
 from Products.ERP5Type.tests.ERP5TypeTestCase import ERP5TypeTestCase
 import random
 import StringIO
 import transaction
+import time
 import unittest
 from ZPublisher.HTTPRequest import HTTPRequest
 from ZPublisher.HTTPResponse import HTTPResponse
@@ -368,6 +370,94 @@ class TestERP5JSONWebTokenPlugin(ERP5TypeTestCase):
       }
     )
 
+  def test_expiration_delay(self):
+    """
+    Test an expiration delay.
+    """
+    password = "%s" % random.random()
+    person = self.person = self._createPerson(
+      self.new_id,
+      password=password,
+      )
+    self.tic()
+    request = self.do_fake_request(
+      "GET",
+      {"HTTP_AUTHORIZATION": "Basic " + base64.b64encode("%s:%s" % (
+        person.getReference(), password))})
+    self.portal.acl_users[self.test_id].manage_setERP5JSONWebTokenPluginExtpirationDelay(2)
+    ret = self.portal.acl_users[self.test_id].extractCredentials(request)
+    ret = self.portal.acl_users[self.test_id].authenticateCredentials(ret)
+    response_cookie_dict = self.REQUEST.response.cookies
+    erp5_jwt_cookie = response_cookie_dict.get('erp5_jwt')
+    request = self.do_fake_request("GET")
+    request.cookies['erp5_jwt'] = erp5_jwt_cookie['value']
+    ret = self.portal.acl_users[self.test_id].extractCredentials(request)
+    self.assertEquals(ret,
+      {
+        'person_relative_url': person.getRelativeUrl(),
+        'remote_host': 'bobo.remote.host',
+        'remote_address': '204.183.226.81 '
+      }
+    )
+    time.sleep(3)
+    request = self.do_fake_request("GET")
+    request.cookies['erp5_jwt'] = erp5_jwt_cookie['value']
+    ret = self.portal.acl_users[self.test_id].extractCredentials(request)
+    self.assertIsNone(ret)
+
+  def test_expiration_delay_deactivated_by_default(self):
+    """
+    Test an expiration delay is deactivated by default
+    """
+    password = "%s" % random.random()
+    person = self.person = self._createPerson(
+      self.new_id,
+      password=password,
+      )
+    self.tic()
+    request = self.do_fake_request(
+      "GET",
+      {"HTTP_AUTHORIZATION": "Basic " + base64.b64encode("%s:%s" % (
+        person.getReference(), password))})
+    ret = self.portal.acl_users[self.test_id].extractCredentials(request)
+    ret = self.portal.acl_users[self.test_id].authenticateCredentials(ret)
+    response_cookie_dict = self.REQUEST.response.cookies
+    erp5_jwt_cookie = response_cookie_dict.get('erp5_jwt')
+    decoded_value = jwt.decode(erp5_jwt_cookie["value"], verify=False)
+    self.assertTrue("exp" not in decoded_value)
+
+  def test_expiration_delay_deactivated_when_set_to_0(self):
+    """
+    Test an expiration delay is deactivated by default
+    """
+    password = "%s" % random.random()
+    person = self.person = self._createPerson(
+      self.new_id,
+      password=password,
+      )
+    self.tic()
+    self.portal.acl_users[self.test_id].manage_setERP5JSONWebTokenPluginExtpirationDelay(2)
+    request = self.do_fake_request(
+      "GET",
+      {"HTTP_AUTHORIZATION": "Basic " + base64.b64encode("%s:%s" % (
+        person.getReference(), password))})
+    ret = self.portal.acl_users[self.test_id].extractCredentials(request)
+    ret = self.portal.acl_users[self.test_id].authenticateCredentials(ret)
+    response_cookie_dict = self.REQUEST.response.cookies
+    erp5_jwt_cookie = response_cookie_dict.get('erp5_jwt')
+    decoded_value = jwt.decode(erp5_jwt_cookie["value"], verify=False)
+    self.assertTrue("exp" in decoded_value)
+    self.portal.acl_users[self.test_id].manage_setERP5JSONWebTokenPluginExtpirationDelay(0)
+    request = self.do_fake_request(
+      "GET",
+      {"HTTP_AUTHORIZATION": "Basic " + base64.b64encode("%s:%s" % (
+        person.getReference(), password))})
+    ret = self.portal.acl_users[self.test_id].extractCredentials(request)
+    ret = self.portal.acl_users[self.test_id].authenticateCredentials(ret)
+    response_cookie_dict = self.REQUEST.response.cookies
+    erp5_jwt_cookie = response_cookie_dict.get('erp5_jwt')
+    decoded_value = jwt.decode(erp5_jwt_cookie["value"], verify=False)
+    self.assertTrue("exp" not in decoded_value)
 
 def test_suite():
   suite = unittest.TestSuite()

product/ERP5Security/www/ERP5Security_setERP5JSONWebTokenPluginExtpirationDelay.zpt

+<h1 tal:replace="structure context/manage_page_header">PAGE HEADER</h1>
+<h2 tal:replace="structure here/manage_tabs"> TABS </h2>
+<h2 tal:define="form_title string:Set ERP5 JSON Web Token Plugin Expiration Delay"
+    tal:replace="structure context/manage_form_title">FORM TITLE</h2>
+
+<p class="form-help">Please input the expiration delay in seconds of the token.
+ The value 0 will deactivate it.</p>
+
+<form action="manage_setERP5JSONWebTokenPluginExtpirationDelay" method="POST">
+
+<table tal:define="expiration_delay request/expiration_delay|context/expiration_delay|python: 0;">
+
+<tr>
+   <td>Expiration delay in seconds of the Token. Deactivated if set to 0</td>
+   <td>
+     <input type="number" name="expiration_delay" value="0"
+            tal:attributes="value expiration_delay;" />
+   </td>
+</tr>
+<tr>
+   <td colspan="2">
+    <input type="submit" value="save"/>
+   </td>
+</tr>
+
+</table>
+
+</form>
+
+<h1 tal:replace="structure context/manage_page_footer">PAGE FOOTER</h1>