feat(api): allow legacy auth from api

api/libretime_api/auth/tests/test_views.py

@@ -1,6 +1,16 @@
+from datetime import datetime
+from http.cookies import SimpleCookie
+
 import pytest
 from rest_framework.test import APIClient
 
+from ...legacy.models import (
+    LEGACY_SESSION_LIFETIME,
+    LegacySession,
+    legacy_session_encode,
+)
+from ...legacy.tests.models_test import make_legacy_session_data
+
 
 @pytest.mark.django_db
 def test_auth_login(api_client: APIClient):
@@ -12,6 +22,30 @@ def test_auth_login(api_client: APIClient):
     assert response.status_code == 204
     assert "sessionid" in response.cookies
     assert response.cookies["sessionid"].value != ""
+    assert "PHPSESSID" not in response.cookies
+
+
+@pytest.mark.django_db
+def test_auth_login_with_legacy(api_client: APIClient):
+    legacy_session = LegacySession.objects.create(
+        id="jikjr9dlrjl9b0jn9r6tnci47a",
+        modified=datetime.now().timestamp(),
+        lifetime=LEGACY_SESSION_LIFETIME,
+        data=legacy_session_encode(make_legacy_session_data()),
+    )
+
+    api_client.cookies = SimpleCookie({"PHPSESSID": legacy_session.id})
+    response = api_client.post(
+        "/api/v2/auth/login/",
+        {"username": "admin", "password": "admin"},
+        format="json",
+    )
+    assert response.status_code == 204
+    assert "sessionid" in response.cookies
+    assert response.cookies["sessionid"].value != ""
+    assert "PHPSESSID" in response.cookies
+    assert response.cookies["PHPSESSID"].value != ""
+    assert response.cookies["PHPSESSID"].value != legacy_session.id
 
 
 @pytest.mark.django_db
@@ -33,3 +67,24 @@ def test_auth_logout(api_client: APIClient):
     assert response.status_code == 200
     assert "sessionid" in response.cookies
     assert response.cookies["sessionid"].value == ""
+    assert "PHPSESSID" not in response.cookies
+
+
+@pytest.mark.django_db
+def test_auth_logout_with_legacy(api_client: APIClient):
+    api_client.login(username="admin", password="admin")
+
+    legacy_session = LegacySession.objects.create(
+        id="jikjr9dlrjl9b0jn9r6tnci47a",
+        modified=datetime.now().timestamp(),
+        lifetime=LEGACY_SESSION_LIFETIME,
+        data=legacy_session_encode(make_legacy_session_data()),
+    )
+    api_client.cookies["PHPSESSID"] = legacy_session.id
+
+    response = api_client.post("/api/v2/auth/logout/")
+    assert response.status_code == 200
+    assert "sessionid" in response.cookies
+    assert response.cookies["sessionid"].value == ""
+    assert "PHPSESSID" in response.cookies
+    assert response.cookies["PHPSESSID"].value == ""

api/libretime_api/auth/views.py

@@ -1,11 +1,29 @@
 from dj_rest_auth.views import LoginView as BaseLoginView, LogoutView as BaseLogoutView
+from django.conf import settings
+from rest_framework.request import Request
+
+from ..legacy.models import LegacySession
 
 
 class LoginView(BaseLoginView):
-    def post(self, request, *args, **kwargs):
+    def post(self, request: Request, *args, **kwargs):
         response = super().post(request, *args, **kwargs)
 
-        # TODO: Create legacy session
+        legacy_session_id = request.COOKIES.get("PHPSESSID")
+        if legacy_session_id:
+            legacy_session = LegacySession.login(
+                legacy_session_id,
+                request.user,  # type: ignore
+            )
+            if legacy_session is not None:
+                response.set_cookie(
+                    key="PHPSESSID",
+                    value=legacy_session.id,
+                    expires="Session",
+                    samesite="Strict",
+                    httponly=True,
+                    secure=settings.CONFIG.general.public_url.startswith("https://"),
+                )
 
         return response
 
@@ -14,9 +32,12 @@ class LogoutView(BaseLogoutView):
     # Fix schema generation
     serializer_class = None
 
-    def post(self, request, *args, **kwargs):
+    def post(self, request: Request, *args, **kwargs):
         response = super().post(request, *args, **kwargs)
 
-        # TODO: Delete legacy session
+        legacy_session_id = request.COOKIES.get("PHPSESSID")
+        if legacy_session_id:
+            LegacySession.logout(legacy_session_id)
+            response.delete_cookie(key="PHPSESSID", samesite="Lax")
 
         return response

api/libretime_api/legacy/models.py

@@ -0,0 +1,98 @@
+import logging
+from datetime import datetime
+from typing import Optional
+
+from django.db import models
+from django.utils.crypto import get_random_string
+
+from ..core.models import User
+from .vendor import phpserialize
+
+logger = logging.getLogger(__name__)
+
+
+LEGACY_SESSION_LIFETIME = 1440
+
+
+def legacy_session_decode(value: Optional[str]) -> dict:
+    if not value:
+        return {}
+    return phpserialize.loads(value, object_hook=phpserialize.phpobject)
+
+
+def legacy_session_encode(data: Optional[dict]) -> str:
+    if not data:
+        return ""
+    return phpserialize.dumps(data, object_hook=phpserialize.phpobject)
+
+
+class LegacySession(models.Model):
+    id = models.CharField(primary_key=True, max_length=32)
+    modified = models.IntegerField(blank=True, null=True)
+    lifetime = models.IntegerField(blank=True, null=True)
+    data = models.TextField(blank=True, null=True)
+
+    class Meta:
+        managed = False
+        db_table = "sessions"
+
+    @classmethod
+    def login(cls, old_session_id: str, user: User) -> Optional["LegacySession"]:
+        try:
+            old_session = cls.objects.get(id=old_session_id)
+        except cls.DoesNotExist:
+            return None
+
+        # Check session expiration time
+        old_session_expires = (old_session.modified or 0) + (old_session.lifetime or 0)
+        if old_session_expires < datetime.now().timestamp():
+            old_session.delete()
+            return None
+
+        session_data = legacy_session_decode(old_session.data)
+
+        def _datetime_format(value: Optional[datetime]) -> Optional[str]:
+            return value.strftime("%Y-%m-%d %H:%M:%S.%f") if value else None
+
+        user_data = phpserialize.phpobject(
+            name="stdClass",
+            d={
+                "id": user.id,
+                "login": user.username,
+                "email": user.email,
+                "type": user.role,
+                "first_name": user.first_name,
+                "last_name": user.last_name,
+                "lastlogin": _datetime_format(user.last_login),
+                "lastfail": _datetime_format(user.last_failed_login),
+                "login_attempts": user.login_attempts,
+                "skype_contact": user.skype,
+                "jabber_contact": user.jabber,
+                "cell_phone": user.phone,
+            },
+        )
+        if session_data["libretime"]["storage"] is not None:
+            logger.warning("overwriting data in legacy session")
+
+        session_data["libretime"]["storage"] = user_data
+
+        # https://github.com/zf1s/zend-session/blob/f33edeaabeda8def65e18cfb13be2c12664f03c3/library/Zend/Session.php#L532-L541
+        new_session_id = get_random_string(
+            length=26,
+            allowed_chars="0123456789abcdef",
+        )
+        new_session = cls(
+            id=new_session_id,
+            modified=datetime.now().timestamp(),
+            lifetime=LEGACY_SESSION_LIFETIME,
+            data=legacy_session_encode(session_data),
+        )
+
+        old_session.delete()
+        new_session.save()
+
+        return new_session
+
+    @classmethod
+    def logout(cls, old_session_id: str) -> None:
+        cls.objects.filter(id=old_session_id).delete()

api/libretime_api/legacy/tests/models_test.py

@@ -0,0 +1,130 @@
+from datetime import datetime
+from typing import Optional
+
+import pytest
+
+from ...core.models import Role, User
+from ..models import (
+    LEGACY_SESSION_LIFETIME,
+    LegacySession,
+    legacy_session_decode,
+    legacy_session_encode,
+)
+from ..vendor import phpserialize
+
+
+def make_legacy_session_data(session_data: Optional[dict] = None):
+    if session_data is not None:
+        storage = phpserialize.phpobject(name="stdClass", d=session_data)
+    else:
+        storage = None
+
+    return {
+        "__ZF": {"csrf_namespace": {"ENT": 1686060245}},
+        "csrf_namespace": {"authtoken": "985a17a922c86a12f28c3ebd1abd09375e1dfde4"},
+        "libretime": {"storage": storage},
+    }
+
+
+TEST_SESSION_DATA_RAW = """a:3:{s:4:"__ZF";a:1:{s:14:"csrf_namespace";a:1:{s:3:"ENT";i:1686060245;}}s:14:"csrf_namespace";a:1:{s:9:"authtoken";s:40:"985a17a922c86a12f28c3ebd1abd09375e1dfde4";}s:9:"libretime";a:1:{s:7:"storage";O:8:"stdClass":13:{s:2:"id";i:1;s:5:"login";s:5:"admin";s:4:"pass";s:32:"21232f297a57a5a743894a0e4a801fc3";s:4:"type";s:1:"A";s:10:"first_name";s:0:"";s:9:"last_name";s:0:"";s:9:"lastlogin";N;s:8:"lastfail";N;s:13:"skype_contact";N;s:14:"jabber_contact";N;s:5:"email";N;s:10:"cell_phone";N;s:14:"login_attempts";i:0;}}}"""  # pylint: disable=line-too-long
+TEST_SESSION_DATA = make_legacy_session_data(
+    {
+        "id": 1,
+        "login": "admin",
+        "pass": "21232f297a57a5a743894a0e4a801fc3",
+        "type": "A",
+        "first_name": "",
+        "last_name": "",
+        "lastlogin": None,
+        "lastfail": None,
+        "skype_contact": None,
+        "jabber_contact": None,
+        "email": None,
+        "cell_phone": None,
+        "login_attempts": 0,
+    }
+)
+
+
+def test_legacy_session_decode():
+    assert legacy_session_decode(TEST_SESSION_DATA_RAW) == TEST_SESSION_DATA
+
+
+def test_legacy_session_encode():
+    print(legacy_session_encode(TEST_SESSION_DATA))
+    assert legacy_session_encode(TEST_SESSION_DATA) == TEST_SESSION_DATA_RAW
+
+
+@pytest.fixture(name="old_session")
+def old_session_fixture():
+    return LegacySession.objects.create(
+        id="jikjr9dlrjl9b0jn9r6tnci47a",
+        modified=datetime.now().timestamp(),
+        lifetime=LEGACY_SESSION_LIFETIME,
+        data=legacy_session_encode(make_legacy_session_data()),
+    )
+
+
+@pytest.mark.django_db
+def test_legacy_session_login(old_session: LegacySession):
+    user = User.objects.create_user(
+        role=Role.HOST,
+        username="test",
+        password="test",
+        email="test@example.com",
+        first_name="test",
+        last_name="user",
+    )
+
+    new_session = LegacySession.login(old_session.id, user)
+    assert new_session is not None
+    assert not LegacySession.objects.filter(id=old_session.id).exists()
+    assert old_session.id != new_session.id
+
+    new_session_data = legacy_session_decode(new_session.data)
+    assert new_session_data == make_legacy_session_data(
+        {
+            "id": user.id,
+            "login": "test",
+            "email": "test@example.com",
+            "type": "H",
+            "first_name": "test",
+            "last_name": "user",
+            "lastlogin": None,
+            "lastfail": None,
+            "login_attempts": None,
+            "skype_contact": None,
+            "jabber_contact": None,
+            "cell_phone": None,
+        },
+    )
+
+
+@pytest.mark.django_db
+def test_legacy_session_login_expired():
+    user = User.objects.create_user(
+        role=Role.HOST,
+        username="test",
+        password="test",
+        email="test@example.com",
+        first_name="test",
+        last_name="user",
+    )
+
+    old_session = LegacySession.objects.create(
+        id="jikjr9dlrjl9b0jn9r6tnci47a",
+        modified=datetime.now().timestamp() - LEGACY_SESSION_LIFETIME - 1,
+        lifetime=LEGACY_SESSION_LIFETIME,
+        data=legacy_session_encode(make_legacy_session_data()),
+    )
+
+    new_session = LegacySession.login(old_session.id, user)
+
+    assert new_session is None
+    assert not LegacySession.objects.filter(id=old_session.id).exists()
+
+
+@pytest.mark.django_db
+def test_legacy_session_logout(old_session: LegacySession):
+    LegacySession.logout(old_session.id)
+    assert not LegacySession.objects.filter(id=old_session.id).exists()

api/libretime_api/legacy/vendor/phpserialize.py

@@ -60,7 +60,16 @@ def __setattr__(self, name, value):
         self.__php_vars__[name] = value
 
     def __repr__(self):
-        return f"<phpobject {self.__name__!r}>"
+        return f"<phpobject name={self.__name__!r} d={self._asdict()!r}>"
+
+    def __eq__(self, value: object) -> bool:
+        if not isinstance(value, phpobject):
+            return False
+
+        if self.__name__ != value.__name__:
+            return False
+
+        return self._asdict() == value._asdict()
 
 
 def convert_member_dict(d):