Lemur: Fix API key expiration bugs

lemur/api_keys/cli.py

@@ -10,7 +10,7 @@
 from lemur.api_keys import service as api_key_service
 from lemur.auth.service import create_token
 
-from datetime import datetime
+import time
 
 manager = Manager(usage="Handles all api key related tasks.")
 
@@ -31,8 +31,8 @@ def create(uid, name, ttl):
     key = api_key_service.create(
         user_id=uid,
         name=name,
-        ttl=ttl,
-        issued_at=int(datetime.utcnow().timestamp()),
+        ttl=int(ttl),
+        issued_at=int(time.time()),
         revoked=False,
     )
     print("[+] Successfully created a new api key. Generating a JWT...")

lemur/api_keys/schemas.py

@@ -7,12 +7,18 @@
 """
 
 from flask import g
-from marshmallow import fields
+from marshmallow import fields, validates, ValidationError
 
 from lemur.common.schema import LemurInputSchema, LemurOutputSchema
 from lemur.users.schemas import UserNestedOutputSchema, UserInputSchema
 
 
+def _validate_ttl(value):
+    """TTL must be -1 (infinite) or a positive integer (days)."""
+    if value is not None and (value == 0 or value < -1):
+        raise ValidationError("TTL must be -1 (infinite) or a positive integer (days).")
+
+
 def current_user_id():
     return {
         "id": g.current_user.id,
@@ -28,6 +34,10 @@ class ApiKeyInputSchema(LemurInputSchema):
     )
     ttl = fields.Integer()
 
+    @validates("ttl")
+    def validate_ttl(self, value):
+        _validate_ttl(value)
+
 
 class ApiKeyRevokeSchema(LemurInputSchema):
     id = fields.Integer(required=True)
@@ -37,11 +47,19 @@ class ApiKeyRevokeSchema(LemurInputSchema):
     ttl = fields.Integer()
     issued_at = fields.Integer(required=False)
 
+    @validates("ttl")
+    def validate_ttl(self, value):
+        _validate_ttl(value)
+
 
 class UserApiKeyInputSchema(LemurInputSchema):
     name = fields.String(required=False)
     ttl = fields.Integer()
 
+    @validates("ttl")
+    def validate_ttl(self, value):
+        _validate_ttl(value)
+
 
 class ApiKeyOutputSchema(LemurOutputSchema):
     jwt = fields.String()

lemur/api_keys/views.py

@@ -8,7 +8,7 @@
 
 """
 
-from datetime import datetime
+import time
 
 from flask import Blueprint, g
 from flask_restful import reqparse, Api
@@ -147,7 +147,7 @@ def post(self, data=None):
             user_id=data["user"]["id"],
             ttl=data["ttl"],
             revoked=False,
-            issued_at=int(datetime.utcnow().timestamp()),
+            issued_at=int(time.time()),
         )
         return dict(
             jwt=create_token(access_token.user_id, access_token.id, access_token.ttl)
@@ -267,7 +267,7 @@ def post(self, user_id, data=None):
             user_id=user_id,
             ttl=data["ttl"],
             revoked=False,
-            issued_at=int(datetime.utcnow().timestamp()),
+            issued_at=int(time.time()),
         )
         return dict(
             jwt=create_token(access_token.user_id, access_token.id, access_token.ttl)

lemur/auth/service.py

@@ -107,9 +107,15 @@ def decorated_function(*args, **kwargs):
                 token,
                 current_app.config["LEMUR_TOKEN_SECRET"],
                 algorithms=[header_data["alg"]],
-                # Disable strict 'sub' validation to support both old tokens (int sub)
-                # and new tokens (string sub) during migration to PyJWT 2.10+
-                options={"verify_sub": False},
+                options={
+                    # Disable strict 'sub' validation to support both old tokens (int sub)
+                    # and new tokens (string sub) during migration to PyJWT 2.10+
+                    "verify_sub": False,
+                    # Disable built-in exp verification: API keys with ttl=-1 have no exp
+                    # claim. Expiration is handled manually below for both API keys and
+                    # regular user tokens.
+                    "verify_exp": False,
+                },
             )
         except jwt.DecodeError:
             return dict(message="Token is invalid"), 403
@@ -118,14 +124,20 @@ def decorated_function(*args, **kwargs):
         except jwt.InvalidTokenError:
             return dict(message="Token is invalid"), 403
 
+        # Manual exp check for regular user tokens (since we disabled verify_exp
+        # in jwt.decode to support API keys without exp claims).
+        if "exp" in payload and "aid" not in payload:
+            if datetime.utcnow() >= datetime.utcfromtimestamp(payload["exp"]):
+                return dict(message="Token has expired"), 403
+
         if "aid" in payload:
             access_key = api_key_service.get(payload["aid"])
             if access_key.revoked:
                 return dict(message="Token has been revoked"), 403
             if access_key.ttl != -1:
                 current_time = datetime.utcnow()
                 # API key uses days
-                expired_time = datetime.fromtimestamp(access_key.issued_at) + timedelta(
+                expired_time = datetime.utcfromtimestamp(access_key.issued_at) + timedelta(
                     days=access_key.ttl
                 )
                 if current_time >= expired_time:

lemur/tests/test_api_keys.py

@@ -100,6 +100,38 @@ def test_api_key_list_post_valid_no_permission(client, token, status):
     )
 
 
+@pytest.mark.parametrize(
+    "token,ttl,status",
+    [
+        (VALID_USER_HEADER_TOKEN, 0, 400),
+        (VALID_USER_HEADER_TOKEN, -2, 400),
+        (VALID_USER_HEADER_TOKEN, -100, 400),
+        (VALID_ADMIN_HEADER_TOKEN, 0, 400),
+        (VALID_USER_HEADER_TOKEN, -1, 200),
+        (VALID_USER_HEADER_TOKEN, 30, 200),
+    ],
+)
+def test_api_key_list_post_ttl_validation(client, token, ttl, status):
+    assert (
+        client.post(
+            api.url_for(ApiKeyList),
+            data=json.dumps(
+                {
+                    "name": "ttl validation test",
+                    "user": {
+                        "id": 1,
+                        "username": "example",
+                        "email": "example@test.net",
+                    },
+                    "ttl": ttl,
+                }
+            ),
+            headers=token,
+        ).status_code
+        == status
+    )
+
+
 @pytest.mark.parametrize(
     "token,status",
     [

setup.py

@@ -81,7 +81,7 @@ def make_release_tree(self, *a, **kw):
         with open(os.path.join(dist_path, "lemur-package.json"), "w") as fp:
             json.dump(
                 {
-                    "createdAt": datetime.datetime.utcnow().isoformat() + "Z",
+                    "createdAt": datetime.datetime.now(datetime.timezone.utc).isoformat(),
                 },
                 fp,
             )