Add endpoint for JWT refresh tokens

Author: stveit

changelog.d/3270.added.md

@@ -0,0 +1 @@
+Add endpoint for JWT refresh tokens

python/nav/web/api/v1/urls.py

@@ -73,4 +73,5 @@
         name="prefix-usage-detail",
     ),
     re_path(r'^', include(router.urls)),
+    re_path(r'^refresh/$', views.JWTRefreshViewSet.as_view(), name='jwt-refresh'),
 ]

python/nav/web/api/v1/views.py

@@ -45,13 +45,21 @@
 from oidc_auth.authentication import JSONWebTokenAuthentication
 
 from nav.models import manage, event, cabling, rack, profiles
+from nav.models.api import JWTRefreshToken
 from nav.models.fields import INFINITY, UNRESOLVED
 from nav.web.servicecheckers import load_checker_classes
 from nav.util import auth_token, is_valid_cidr
 
 from nav.buildconf import VERSION
 from nav.web.api.v1 import serializers, alert_serializers
 from nav.web.status2 import STATELESS_THRESHOLD
+from nav.web.jwtgen import (
+    generate_access_token,
+    generate_refresh_token,
+    hash_token,
+    decode_token,
+    is_active,
+)
 from nav.macaddress import MacPrefix
 from .auth import (
     APIPermission,
@@ -1153,3 +1161,45 @@ class ModuleViewSet(NAVAPIMixin, viewsets.ReadOnlyModelViewSet):
         'device__serial',
     )
     serializer_class = serializers.ModuleSerializer
+
+
+class JWTRefreshViewSet(APIView):
+    """
+    Accepts a valid refresh token.
+    Returns a new refresh token and an access token.
+    """
+
+    def post(self, request):
+        incoming_token = request.data.get('refresh_token')
+        token_hash = hash_token(incoming_token)
+        try:
+            # If hash exists in the database, then we know it is a real token
+            db_token = JWTRefreshToken.objects.get(hash=token_hash)
+        except JWTRefreshToken.DoesNotExist:
+            return Response("Invalid token", status=status.HTTP_403_FORBIDDEN)
+
+        claims = decode_token(incoming_token)
+        if not is_active(claims['exp'], claims['nbf']):
+            return Response("Inactive token", status=status.HTTP_403_FORBIDDEN)
+
+        if db_token.revoked:
+            return Response(
+                "This token has been revoked", status=status.HTTP_403_FORBIDDEN
+            )
+
+        access_token = generate_access_token(claims)
+        refresh_token = generate_refresh_token(claims)
+
+        new_claims = decode_token(refresh_token)
+        new_hash = hash_token(refresh_token)
+        db_token.hash = new_hash
+        db_token.expires = datetime.fromtimestamp(new_claims['exp'])
+        db_token.activates = datetime.fromtimestamp(new_claims['nbf'])
+        db_token.last_used = datetime.now()
+        db_token.save()
+
+        response_data = {
+            'access_token': access_token,
+            'refresh_token': refresh_token,
+        }
+        return Response(response_data)

python/nav/web/jwtgen.py

@@ -1,5 +1,6 @@
 from datetime import datetime, timedelta, timezone
 from typing import Any, Optional
+import hashlib
 
 import jwt
 
@@ -64,3 +65,15 @@ def is_active(exp: float, nbf: float) -> bool:
     expires = datetime.fromtimestamp(exp, tz=timezone.utc)
     activates = datetime.fromtimestamp(nbf, tz=timezone.utc)
     return now >= activates and now < expires
+
+
+def hash_token(token: str) -> str:
+    """Hashes a token with SHA256"""
+    hash_object = hashlib.sha256(token.encode('utf-8'))
+    hex_dig = hash_object.hexdigest()
+    return hex_dig
+
+
+def decode_token(token: str) -> dict[str, Any]:
+    """Decodes a token in JWT format and returns the data of the decoded token"""
+    return jwt.decode(token, options={'verify_signature': False})

tests/integration/jwt_refresh_endpoint_test.py

@@ -0,0 +1,214 @@
+import hashlib
+from typing import Generator
+
+import jwt
+import pytest
+from datetime import datetime, timedelta, timezone
+
+from unittest.mock import MagicMock, Mock, patch
+
+from django.urls import reverse
+from nav.models.api import JWTRefreshToken
+
+
+def test_token_not_in_database_should_be_rejected(db, api_client, url, active_token):
+    token_hash = hashlib.sha256(active_token.encode('utf-8')).hexdigest()
+    assert not JWTRefreshToken.objects.filter(hash=token_hash).exists()
+    response = api_client.post(
+        url,
+        follow=True,
+        data={
+            'refresh_token': active_token,
+        },
+    )
+    assert response.status_code == 403
+
+
+def test_inactive_token_should_be_rejected(db, api_client, url, inactive_token):
+    now = datetime.now()
+    token_hash = hashlib.sha256(inactive_token.encode('utf-8')).hexdigest()
+    db_token = JWTRefreshToken(
+        name="testtoken",
+        hash=token_hash,
+        expires=now - timedelta(hours=1),
+        activates=now - timedelta(hours=2),
+    )
+    db_token.save()
+
+    response = api_client.post(
+        url,
+        follow=True,
+        data={
+            'refresh_token': inactive_token,
+        },
+    )
+
+    assert response.status_code == 403
+
+
+def test_valid_token_should_be_accepted(db, api_client, url, active_token):
+    data = jwt.decode(active_token, options={'verify_signature': False})
+    token_hash = token_hash = hashlib.sha256(active_token.encode('utf-8')).hexdigest()
+    db_token = JWTRefreshToken(
+        name="testtoken",
+        hash=token_hash,
+        expires=datetime.fromtimestamp(data['exp']),
+        activates=datetime.fromtimestamp(data['nbf']),
+    )
+    db_token.save()
+    response = api_client.post(
+        url,
+        follow=True,
+        data={
+            'refresh_token': active_token,
+        },
+    )
+    assert response.status_code == 200
+
+
+def test_valid_token_should_be_replaced_by_new_token_in_db(
+    db, api_client, url, active_token
+):
+    token_hash = token_hash = hashlib.sha256(active_token.encode('utf-8')).hexdigest()
+    data = jwt.decode(active_token, options={'verify_signature': False})
+    db_token = JWTRefreshToken(
+        name="testtoken",
+        hash=token_hash,
+        expires=datetime.fromtimestamp(data['exp']),
+        activates=datetime.fromtimestamp(data['nbf']),
+    )
+    db_token.save()
+    response = api_client.post(
+        url,
+        follow=True,
+        data={
+            'refresh_token': active_token,
+        },
+    )
+    assert response.status_code == 200
+    assert not JWTRefreshToken.objects.filter(hash=token_hash).exists()
+    new_token = response.data.get("refresh_token")
+    new_hash = hashlib.sha256(new_token.encode('utf-8')).hexdigest()
+    assert JWTRefreshToken.objects.filter(hash=new_hash).exists()
+
+
+def test_should_include_access_and_refresh_token_in_response(
+    db, api_client, url, active_token
+):
+    token_hash = hashlib.sha256(active_token.encode('utf-8')).hexdigest()
+    data = jwt.decode(active_token, options={'verify_signature': False})
+    db_token = JWTRefreshToken(
+        name="testtoken",
+        hash=token_hash,
+        expires=datetime.fromtimestamp(data['exp']),
+        activates=datetime.fromtimestamp(data['nbf']),
+    )
+    db_token.save()
+    response = api_client.post(
+        url,
+        follow=True,
+        data={
+            'refresh_token': active_token,
+        },
+    )
+    assert response.status_code == 200
+    assert "access_token" in response.data
+    assert "refresh_token" in response.data
+
+
+def test_revoked_token_should_be_rejected(db, api_client, url, active_token):
+    token_hash = hashlib.sha256(active_token.encode('utf-8')).hexdigest()
+    data = jwt.decode(active_token, options={'verify_signature': False})
+    db_token = JWTRefreshToken(
+        name="testtoken",
+        hash=token_hash,
+        expires=datetime.fromtimestamp(data['exp']),
+        activates=datetime.fromtimestamp(data['nbf']),
+        revoked=True,
+    )
+    db_token.save()
+    response = api_client.post(
+        url,
+        follow=True,
+        data={
+            'refresh_token': active_token,
+        },
+    )
+    assert response.status_code == 403
+
+
+def test_last_used_should_be_updated_after_token_is_used(
+    db, api_client, url, active_token
+):
+    token_hash = hashlib.sha256(active_token.encode('utf-8')).hexdigest()
+    data = jwt.decode(active_token, options={'verify_signature': False})
+    db_token = JWTRefreshToken(
+        name="testtoken",
+        hash=token_hash,
+        expires=datetime.fromtimestamp(data['exp']),
+        activates=datetime.fromtimestamp(data['nbf']),
+    )
+    db_token.save()
+    assert db_token.last_used is None
+    response = api_client.post(
+        url,
+        follow=True,
+        data={
+            'refresh_token': active_token,
+        },
+    )
+    new_token = response.data.get("refresh_token")
+    new_hash = hashlib.sha256(new_token.encode('utf-8')).hexdigest()
+    assert JWTRefreshToken.objects.get(hash=new_hash).last_used is not None
+
+
+@pytest.fixture()
+def inactive_token(nav_name, rsa_private_key) -> str:
+    now = datetime.now(timezone.utc)
+    claims = {
+        'exp': (now - timedelta(hours=1)).timestamp(),
+        'nbf': (now - timedelta(hours=2)).timestamp(),
+        'iat': (now - timedelta(hours=2)).timestamp(),
+        'aud': nav_name,
+        'iss': nav_name,
+        'token_type': 'refresh_token',
+    }
+    token = jwt.encode(claims, rsa_private_key, algorithm="RS256")
+    return token
+
+
+@pytest.fixture()
+def active_token(nav_name, rsa_private_key) -> str:
+    now = datetime.now(timezone.utc)
+    claims = {
+        'exp': (now + timedelta(hours=1)).timestamp(),
+        'nbf': now.timestamp(),
+        'iat': now.timestamp(),
+        'aud': nav_name,
+        'iss': nav_name,
+        'token_type': 'refresh_token',
+    }
+    token = jwt.encode(claims, rsa_private_key, algorithm="RS256")
+    return token
+
+
+@pytest.fixture()
+def url():
+    return reverse('api:1:jwt-refresh')
+
+
+@pytest.fixture(scope="module", autouse=True)
+def jwtconf_mock(rsa_private_key, nav_name) -> Generator[MagicMock, None, None]:
+    """Mocks the get_nave_name and get_nav_private_key functions for
+    the JWTConf class
+    """
+    with patch("nav.web.jwtgen.JWTConf") as _jwtconf_mock:
+        instance = _jwtconf_mock.return_value
+        instance.get_nav_name = Mock(return_value=nav_name)
+        instance.get_nav_private_key = Mock(return_value=rsa_private_key)
+        yield _jwtconf_mock
+
+
+@pytest.fixture(scope="module")
+def nav_name() -> str:
+    return "nav"

tests/unittests/web/jwtgen_test.py

@@ -1,10 +1,16 @@
+from typing import Any
 import pytest
 from unittest.mock import Mock, patch
 from datetime import datetime
 
 import jwt
 
-from nav.web.jwtgen import generate_access_token, generate_refresh_token
+from nav.web.jwtgen import (
+    generate_access_token,
+    generate_refresh_token,
+    hash_token,
+    decode_token,
+)
 
 
 class TestTokenGeneration:
@@ -55,6 +61,16 @@ def test_token_type_should_be_refresh_token(self):
         assert data['token_type'] == "refresh_token"
 
 
+class TestHashToken:
+    def test_should_return_correct_hash(self, token_string, token_hash):
+        assert hash_token(token_string) == token_hash
+
+
+class TestDecodeToken:
+    def test_should_return_expected_data(self, token_string, token_data):
+        assert decode_token(token_string) == token_data
+
+
 @pytest.fixture(scope="module", autouse=True)
 def jwtconf_mock(rsa_private_key, nav_name) -> str:
     """Mocks the get_nav_name and get_nav_private_key functions for
@@ -70,3 +86,35 @@ def jwtconf_mock(rsa_private_key, nav_name) -> str:
 @pytest.fixture(scope="module")
 def nav_name() -> str:
     yield "nav"
+
+
+@pytest.fixture(scope="module")
+def token_string() -> str:
+    """String representation of a token. Matching data is in `token_data`
+    and expected hash is in `token_hash`
+    """
+    token = (
+        "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpYXQ"
+        "iOjE3NDA0Nzg4NTQsImV4cCI6MTc0MDU2NTI1NH0.2GbcpbwzVOAV7"
+        "nv4lAS_ISrw-g9WKvhKKnpN9dhSL6s"
+    )
+    return token
+
+
+@pytest.fixture(scope="module")
+def token_hash() -> str:
+    """SHA256 hash of a token. Matching data is in `token_data`
+    and the actual token string is in `token_string`
+    """
+    return "91d0d189dde6a7423b884f8bb285b17f9706d21e6d0ce45aac028a22b3067395"
+
+
+@pytest.fixture(scope="module")
+def token_data() -> dict[str, Any]:
+    """Payload of a token. The actual token string is in `token_string`
+    and hash of the token in `token_hash`
+    """
+    return {
+        "iat": 1740478854,
+        "exp": 1740565254,
+    }