1"""Synthetic Jellyfin 12.2 SQLite fixtures; never access a real media database."""
  2
  3import importlib.util
  4import json
  5import os
  6from pathlib import Path
  7import sqlite3
  8import tempfile
  9import unittest
 10from unittest.mock import patch
 11
 12spec = importlib.util.spec_from_file_location("export_auth", Path(__file__).with_name("export-jellyfin-auth.py"))
 13exporter = importlib.util.module_from_spec(spec)
 14spec.loader.exec_module(exporter)
 15FIXTURE = json.loads((Path(__file__).parents[1] / "testdata/offline-auth.json").read_text())
 16
 17
 18class ExportTests(unittest.TestCase):
 19    def setUp(self):
 20        self.temp = tempfile.TemporaryDirectory()
 21        self.addCleanup(self.temp.cleanup)
 22        self.root = Path(self.temp.name)
 23        self.database = self.root / "jellyfin.db"
 24        self.output = self.root / "auth/users.json"
 25        self.db = sqlite3.connect(self.database)
 26        self.addCleanup(self.db.close)
 27        self.db.executescript('''
 28            PRAGMA journal_mode=WAL;
 29            PRAGMA wal_autocheckpoint=0;
 30            CREATE TABLE Users (Id TEXT, Username TEXT, Password TEXT, AuthenticationProviderId TEXT, MustUpdatePassword INTEGER);
 31            CREATE TABLE Permissions (UserId TEXT, Kind INTEGER, Value INTEGER);
 32            CREATE TABLE AccessSchedules (UserId TEXT);
 33        ''')
 34        self.add_user("a", "Alice")
 35
 36    def add_user(self, char, name, provider=exporter.PROVIDER, password="fixture", disabled=0, remote=1, must_update=0):
 37        user_id = char * 32
 38        verifier = FIXTURE["users"][0]["passwordHash"] if password == "fixture" else password
 39        self.db.execute("INSERT INTO Users VALUES (?,?,?,?,?)", (user_id, name, verifier, provider, must_update))
 40        self.db.executemany("INSERT INTO Permissions VALUES (?,?,?)", [(user_id, 2, disabled), (user_id, 4, remote)])
 41        self.db.commit()
 42        return user_id
 43
 44    def export(self):
 45        return exporter.export(self.database, self.output, "2026-01-01T00:00:00Z")
 46
 47    def test_export_sees_committed_wal_and_no_plaintext(self):
 48        # Keep the writer open, forcing the exporter to read committed WAL pages.
 49        self.add_user("c", "Legacy", password=FIXTURE["users"][1]["passwordHash"])
 50        before = self.db.total_changes
 51        enabled, _ = self.export()
 52        doc = json.loads(self.output.read_text())
 53        self.assertEqual(enabled, 2)
 54        self.assertEqual(doc["users"][0]["id"], "a" * 32)
 55        self.assertEqual(doc["users"][0]["passwordHash"], FIXTURE["users"][0]["passwordHash"])
 56        self.assertNotIn("fixture-only", self.output.read_text())
 57        self.assertEqual(self.output.stat().st_mode & 0o777, 0o600)
 58        self.assertEqual(self.db.total_changes, before)
 59        self.assertEqual(self.db.execute("SELECT count(*) FROM Users").fetchone()[0], 2)
 60
 61    def test_blocked_external_passwordless_and_restricted_users(self):
 62        self.add_user("b", "Disabled", disabled=1)
 63        self.add_user("c", "LDAP", provider="External.Provider")
 64        self.add_user("d", "NoPassword", password=None)
 65        self.add_user("e", "LocalOnly", remote=0)
 66        self.add_user("f", "MustChange", must_update=1)
 67        uid = self.add_user("1", "Scheduled")
 68        self.db.execute("INSERT INTO AccessSchedules VALUES (?)", (uid,))
 69        self.db.commit()
 70        enabled, counts = self.export()
 71        self.assertEqual(enabled, 1)
 72        self.assertEqual(counts, {"external": 1, "passwordless": 1, "disabled": 1, "restricted": 3})
 73        doc = json.loads(self.output.read_text())
 74        for user in doc["users"]:
 75            if user["disabled"]:
 76                self.assertEqual(user["passwordHash"], "")
 77
 78    def test_failed_exports_preserve_previous_and_success_creates_backup(self):
 79        self.export()
 80        previous = self.output.read_bytes()
 81        self.db.execute("UPDATE Users SET Password='not-a-hash'")
 82        self.db.commit()
 83        with self.assertRaises(ValueError):
 84            self.export()
 85        self.assertEqual(self.output.read_bytes(), previous)
 86        self.db.execute("UPDATE Users SET Password=?", (FIXTURE["users"][0]["passwordHash"],))
 87        self.db.commit()
 88        self.export()
 89        self.assertEqual(Path(str(self.output) + ".previous").read_bytes(), previous)
 90        self.assertEqual(Path(str(self.output) + ".previous").stat().st_mode & 0o777, 0o600)
 91        with patch.object(exporter.os, "replace", side_effect=OSError("synthetic failure")):
 92            with self.assertRaises(OSError):
 93                self.export()
 94        self.assertEqual(self.output.read_bytes(), previous)
 95
 96    def test_missing_permissions_schema_and_unsafe_destinations(self):
 97        self.db.execute("DELETE FROM Permissions WHERE Kind=2")
 98        self.db.commit()
 99        with self.assertRaises(ValueError):
100            self.export()
101        self.db.execute("DROP TABLE Permissions")
102        self.db.commit()
103        with self.assertRaises(ValueError):
104            self.export()
105        with self.assertRaises(ValueError):
106            exporter.export(self.database, self.database)
107
108    def test_private_file_and_symlink_checks(self):
109        self.export()
110        self.output.chmod(0o644)
111        with self.assertRaises(ValueError):
112            self.export()
113        self.output.chmod(0o600)
114        link = self.root / "linked.json"
115        link.symlink_to(self.output)
116        with self.assertRaises(ValueError):
117            exporter.export(self.database, link)
118
119    def test_zero_accounts_is_valid_and_bounds_reject_dangerous_hashes(self):
120        self.db.execute("DELETE FROM Users")
121        self.db.commit()
122        enabled, _ = self.export()
123        self.assertEqual(enabled, 0)
124        self.assertEqual(json.loads(self.output.read_text())["users"], [])
125        for encoded in ["plaintext", "$PBKDF2-SHA512$iterations=9999999$0011223344556677$" + "00" * 64]:
126            with self.assertRaises(ValueError):
127                exporter.validate_hash(encoded)
128
129
130if __name__ == "__main__":
131    unittest.main()