Rodger/rodgers-ios-music-export
1
0
Fork 0
Code Issues Pull requests Projects Releases 1 Packages Wiki Activity Actions

feat: publish read-only iOS music export skill

Browse source
This commit is contained in:
Rodger Werkhoven 2026-08-10 14:25:24 +02:00
commit 45ade372f6
15 changed files with 2872 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

592
tests/test_ios_music_export.py Normal file
View file

@ -0,0 +1,592 @@
import hashlib
import importlib.util
import os
from pathlib import Path
import sqlite3
import struct
import tempfile
import unittest
from unittest import mock
PLUGIN_ROOT = Path(__file__).resolve().parents[1]
SCRIPT_PATH = (
PLUGIN_ROOT
/ "skills"
/ "rodgers-ios-music-export"
/ "scripts"
/ "ios_music_export.py"
)
SPEC = importlib.util.spec_from_file_location("ios_music_export", SCRIPT_PATH)
MODULE = importlib.util.module_from_spec(SPEC)
SPEC.loader.exec_module(MODULE)
def create_media_database(path):
connection = sqlite3.connect(path)
connection.executescript(
"""
CREATE TABLE item (
item_pid INTEGER PRIMARY KEY,
genre_id INTEGER,
item_artist_pid INTEGER,
album_pid INTEGER,
base_location_id INTEGER
);
CREATE TABLE item_extra (
item_pid INTEGER PRIMARY KEY,
title TEXT,
location TEXT,
file_size INTEGER
);
CREATE TABLE genre (genre_id INTEGER PRIMARY KEY, genre TEXT);
CREATE TABLE item_artist (item_artist_pid INTEGER PRIMARY KEY, item_artist TEXT);
CREATE TABLE album (album_pid INTEGER PRIMARY KEY, album TEXT);
CREATE TABLE base_location (base_location_id INTEGER PRIMARY KEY, path TEXT);
"""
)
connection.executemany(
"INSERT INTO genre VALUES (?, ?)",
[(1, "Electro"), (2, "electro"), (3, "Elektronisch"), (4, "Electro House")],
)
connection.execute("INSERT INTO item_artist VALUES (10, 'Test/Artist')")
connection.execute("INSERT INTO album VALUES (20, 'Test Album')")
connection.execute("INSERT INTO base_location VALUES (30, 'iTunes_Control/Music')")
rows = [
(101, 1, "Same: Track", "F00/AAAA.m4a", 5),
(102, 2, "Same: Track", "F01/BBBB.M4A", 6),
(103, 3, "Andere track", "F02/CCCC.m4a", 7),
(104, 4, "House track", "F03/DDDD.m4a", 8),
]
for item_pid, genre_id, title, location, size in rows:
connection.execute("INSERT INTO item VALUES (?, ?, 10, 20, 30)", (item_pid, genre_id))
connection.execute(
"INSERT INTO item_extra VALUES (?, ?, ?, ?)",
(item_pid, title, location, size),
)
connection.commit()
connection.close()
class FakeAFCClient:
def __init__(self, content):
self.content = content
self.download_calls = []
def download(self, remote_path, local_path, destination_fd=None):
self.download_calls.append(remote_path)
data = self.content[remote_path]
flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL
if hasattr(os, "O_NOFOLLOW"):
flags |= os.O_NOFOLLOW
descriptor = os.open(local_path, flags, 0o600, dir_fd=destination_fd)
with os.fdopen(descriptor, "wb") as output:
output.write(data)
return len(data), hashlib.sha256(data).hexdigest()
def get_file_info(self, remote_path):
if remote_path not in self.content:
raise MODULE.AFCError(MODULE.AFC_E_OBJECT_NOT_FOUND, MODULE.AFC_OP_GET_FILE_INFO)
data = self.content[remote_path]
return {
"st_ifmt": "S_IFREG",
"st_size": str(len(data)),
"st_mtime": "1",
"st_blocks": "1",
}
def sample_plan():
return [
{
"item_pid": "1",
"genre": "Electro",
"source_device_fingerprint": "a" * 64,
"database_snapshot_sha256": "b" * 64,
"title": "One",
"artist": "Artist",
"album": "Album",
"file_size": 5,
"source_mtime": "1",
"source_blocks": "1",
"remote_path": "/iTunes_Control/Music/F00/one.m4a",
"filename": "Artist — One.m4a",
},
{
"item_pid": "2",
"genre": "Electro",
"source_device_fingerprint": "a" * 64,
"database_snapshot_sha256": "b" * 64,
"title": "Two",
"artist": "Artist",
"album": "Album",
"file_size": 6,
"source_mtime": "1",
"source_blocks": "1",
"remote_path": "/iTunes_Control/Music/F01/two.mp3",
"filename": "Artist — Two.mp3",
},
]
class IOSMusicExportTests(unittest.TestCase):
def test_exact_case_insensitive_genre_and_collision_names(self):
with tempfile.TemporaryDirectory() as directory:
database = Path(directory) / "MediaLibrary.sqlitedb"
create_media_database(database)
plan = MODULE.make_export_plan(database, "ELECTRO")
self.assertEqual([row["item_pid"] for row in plan], ["101", "102"])
self.assertEqual(plan[0]["filename"], "Test⁄Artist — Same - Track.m4a")
self.assertEqual(plan[1]["filename"], "Test⁄Artist — Same - Track [2].m4a")
self.assertTrue(plan[0]["remote_path"].endswith("/F00/AAAA.m4a"))
def test_adjacent_genres_are_not_selected(self):
with tempfile.TemporaryDirectory() as directory:
database = Path(directory) / "MediaLibrary.sqlitedb"
create_media_database(database)
plan = MODULE.make_export_plan(database, "Elektronisch")
self.assertEqual(len(plan), 1)
self.assertEqual(plan[0]["item_pid"], "103")
def test_unknown_schema_fails_closed(self):
with tempfile.TemporaryDirectory() as directory:
database = Path(directory) / "MediaLibrary.sqlitedb"
sqlite3.connect(database).close()
with self.assertRaisesRegex(RuntimeError, "unsupported MediaLibrary schema"):
MODULE.make_export_plan(database, "Electro")
def test_two_identical_database_captures_are_consolidated_and_checked(self):
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
source = root / "source.sqlitedb"
create_media_database(source)
client = FakeAFCClient({MODULE.MEDIA_DATABASE_REMOTE: source.read_bytes()})
snapshot = MODULE.snapshot_media_database(client, root / "snapshot-work")
plan = MODULE.make_export_plan(snapshot, "Electro")
self.assertEqual(len(plan), 2)
self.assertEqual(client.download_calls.count(MODULE.MEDIA_DATABASE_REMOTE), 2)
def test_export_hashes_verifies_and_resumes(self):
with tempfile.TemporaryDirectory() as directory:
root = Path(directory).resolve()
destination = root / "music"
manifest = root / "manifest.tsv"
plan = sample_plan()
client = FakeAFCClient({
"/iTunes_Control/Music/F00/one.m4a": b"12345",
"/iTunes_Control/Music/F01/two.mp3": b"abcdef",
})
first = MODULE.export_plan(client, plan, destination, manifest)
verification = MODULE.verify_export(
manifest, destination, "Electro", probe_audio=False
)
second_client = FakeAFCClient({})
second = MODULE.export_plan(second_client, plan, destination, manifest)
self.assertEqual(first["copied_tracks"], 2)
self.assertEqual(verification["manifest_tracks"], 2)
self.assertEqual(verification["hash_errors"], [])
self.assertEqual(second["resumed_tracks"], 2)
self.assertEqual(second_client.download_calls, [])
def test_untrusted_existing_file_is_preserved(self):
with tempfile.TemporaryDirectory() as directory:
destination = Path(directory).resolve() / "music"
destination.mkdir()
(destination / "Artist — One.m4a").write_bytes(b"12345")
plan = [sample_plan()[0]]
with self.assertRaisesRegex(FileExistsError, "preserving untrusted"):
MODULE.inspect_existing_destination(destination, plan, {})
def test_verify_requires_present_nonempty_manifest_and_correct_genre(self):
with tempfile.TemporaryDirectory() as directory:
root = Path(directory).resolve()
destination = root / "music"
destination.mkdir()
manifest = root / "manifest.tsv"
with self.assertRaisesRegex(RuntimeError, "manifest does not exist"):
MODULE.verify_export(manifest, destination, "Electro", probe_audio=False)
MODULE.write_manifest_atomic(manifest, [])
with self.assertRaisesRegex(RuntimeError, "manifest is empty"):
MODULE.verify_export(manifest, destination, "Electro", probe_audio=False)
client = FakeAFCClient({
"/iTunes_Control/Music/F00/one.m4a": b"12345",
"/iTunes_Control/Music/F01/two.mp3": b"abcdef",
})
MODULE.export_plan(client, sample_plan(), destination, manifest)
with self.assertRaisesRegex(RuntimeError, "wrong_genre"):
MODULE.verify_export(manifest, destination, "Jazz", probe_audio=False)
def test_pending_manifest_never_trusts_same_size_final_file(self):
with tempfile.TemporaryDirectory() as directory:
root = Path(directory).resolve()
destination = root / "music"
destination.mkdir()
manifest = root / "manifest.tsv"
row = sample_plan()[0]
(destination / row["filename"]).write_bytes(b"WRONG")
pending = {
**row,
"partial_name": "",
"actual_size": "0",
"sha256": "",
"status": "pending",
}
MODULE.write_manifest_atomic(manifest, [pending])
with self.assertRaisesRegex(RuntimeError, "no trusted source hash"):
MODULE.inspect_existing_destination(destination, [row], MODULE.load_manifest(manifest))
pending["sha256"] = hashlib.sha256(b"WRONG").hexdigest()
MODULE.write_manifest_atomic(manifest, [pending])
with self.assertRaisesRegex(RuntimeError, "status is not complete"):
MODULE.inspect_existing_destination(destination, [row], MODULE.load_manifest(manifest))
def test_late_collision_is_not_overwritten(self):
class RacingClient(FakeAFCClient):
def __init__(self, content, final_path):
super().__init__(content)
self.final_path = final_path
def download(self, remote_path, local_path, destination_fd=None):
self.final_path.write_bytes(b"RIVAL")
return super().download(
remote_path, local_path, destination_fd=destination_fd
)
with tempfile.TemporaryDirectory() as directory:
root = Path(directory).resolve()
destination = root / "music"
manifest = root / "manifest.tsv"
row = sample_plan()[0]
final_path = destination / row["filename"]
client = RacingClient({row["remote_path"]: b"12345"}, final_path)
with self.assertRaisesRegex(RuntimeError, "refusing to replace"):
MODULE.export_plan(client, [row], destination, manifest)
self.assertEqual(final_path.read_bytes(), b"RIVAL")
def test_destination_symlink_swap_cannot_redirect_track_write(self):
with tempfile.TemporaryDirectory() as directory:
root = Path(directory).resolve()
destination = root / "music"
moved_destination = root / "music-original"
escape = root / "escape"
escape.mkdir()
manifest = root / "manifest.tsv"
row = sample_plan()[0]
class SwappingClient(FakeAFCClient):
def download(self, remote_path, local_path, destination_fd=None):
destination.rename(moved_destination)
destination.symlink_to(escape, target_is_directory=True)
return super().download(
remote_path, local_path, destination_fd=destination_fd
)
client = SwappingClient({row["remote_path"]: b"12345"})
with self.assertRaisesRegex(RuntimeError, "destination path changed concurrently"):
MODULE.export_plan(client, [row], destination, manifest)
self.assertFalse((escape / row["filename"]).exists())
self.assertEqual((moved_destination / row["filename"]).read_bytes(), b"12345")
self.assertEqual(MODULE.load_manifest(manifest)[row["filename"]]["status"], "pending")
def test_download_failure_preserves_later_resume_records(self):
class FailingClient(FakeAFCClient):
def download(self, remote_path, local_path, destination_fd=None):
raise IOError("injected download failure")
with tempfile.TemporaryDirectory() as directory:
root = Path(directory).resolve()
destination = root / "music"
manifest = root / "manifest.tsv"
plan = sample_plan()
content = {
"/iTunes_Control/Music/F00/one.m4a": b"12345",
"/iTunes_Control/Music/F01/two.mp3": b"abcdef",
}
MODULE.export_plan(FakeAFCClient(content), plan, destination, manifest)
(destination / plan[0]["filename"]).unlink()
with self.assertRaisesRegex(IOError, "injected"):
MODULE.export_plan(FailingClient(content), plan, destination, manifest)
records = MODULE.load_manifest(manifest)
self.assertEqual(set(records), {plan[0]["filename"], plan[1]["filename"]})
self.assertTrue(records[plan[1]["filename"]]["sha256"])
summary = MODULE.export_plan(FakeAFCClient(content), plan, destination, manifest)
self.assertEqual(summary["copied_tracks"], 1)
self.assertEqual(summary["resumed_tracks"], 1)
def test_interrupted_direct_final_is_preserved_and_not_trusted(self):
class PartialFailureClient(FakeAFCClient):
def download(self, remote_path, local_path, destination_fd=None):
flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL
if hasattr(os, "O_NOFOLLOW"):
flags |= os.O_NOFOLLOW
descriptor = os.open(
local_path, flags, 0o600, dir_fd=destination_fd
)
with os.fdopen(descriptor, "wb") as output:
output.write(b"12")
raise IOError("injected mid-track interruption")
with tempfile.TemporaryDirectory() as directory:
root = Path(directory).resolve()
destination = root / "music"
manifest = root / "manifest.tsv"
row = sample_plan()[0]
client = PartialFailureClient({row["remote_path"]: b"12345"})
with self.assertRaisesRegex(IOError, "mid-track interruption"):
MODULE.export_plan(client, [row], destination, manifest)
final_path = destination / row["filename"]
self.assertEqual(final_path.read_bytes(), b"12")
with self.assertRaisesRegex(RuntimeError, "no trusted source hash"):
MODULE.export_plan(FakeAFCClient({row["remote_path"]: b"12345"}), [row], destination, manifest)
self.assertEqual(final_path.read_bytes(), b"12")
def test_ready_final_is_hash_verified_and_recovered(self):
with tempfile.TemporaryDirectory() as directory:
root = Path(directory).resolve()
destination = root / "music"
destination.mkdir()
manifest = root / "manifest.tsv"
row = sample_plan()[0]
final_path = destination / row["filename"]
final_path.write_bytes(b"12345")
ready = {
**row,
"partial_name": "",
"actual_size": "5",
"sha256": hashlib.sha256(b"12345").hexdigest(),
"status": "ready_to_commit",
}
MODULE.write_manifest_atomic(manifest, [ready])
summary = MODULE.export_plan(FakeAFCClient({}), [row], destination, manifest)
verification = MODULE.verify_export(
manifest, destination, "Electro", probe_audio=False
)
self.assertEqual(summary["recovered_tracks"], 1)
self.assertEqual(verification["hash_errors"], [])
self.assertEqual(final_path.read_bytes(), b"12345")
def test_unrelated_manifest_is_preserved(self):
with tempfile.TemporaryDirectory() as directory:
root = Path(directory).resolve()
destination = root / "music"
manifest = root / "manifest.tsv"
unrelated = {
**sample_plan()[0],
"filename": "Unrelated.m4a",
"partial_name": "",
"actual_size": "5",
"sha256": "c" * 64,
"status": "copied",
}
MODULE.write_manifest_atomic(manifest, [unrelated])
original = manifest.read_bytes()
with self.assertRaisesRegex(RuntimeError, "another export plan"):
MODULE.export_plan(
FakeAFCClient({sample_plan()[0]["remote_path"]: b"12345"}),
[sample_plan()[0]],
destination,
manifest,
)
self.assertEqual(manifest.read_bytes(), original)
self.assertFalse(destination.exists())
def test_late_unrelated_manifest_collision_is_preserved(self):
with tempfile.TemporaryDirectory() as directory:
root = Path(directory).resolve()
destination = root / "music"
manifest = root / "manifest.tsv"
row = sample_plan()[0]
unrelated = {
**row,
"filename": "Late unrelated.m4a",
"partial_name": "",
"actual_size": "5",
"sha256": "c" * 64,
"status": "copied",
}
original_inspect = MODULE.inspect_existing_destination
collided_bytes = []
def inject_manifest_collision(
target, plan, previous_records, destination_descriptor=None
):
result = original_inspect(
target,
plan,
previous_records,
destination_descriptor=destination_descriptor,
)
MODULE.write_manifest_atomic(manifest, [unrelated])
with manifest.open("ab") as output:
output.write(b"torn-unrelated-tail")
collided_bytes.append(manifest.read_bytes())
return result
with mock.patch.object(
MODULE, "inspect_existing_destination", side_effect=inject_manifest_collision
):
with self.assertRaisesRegex(RuntimeError, "another export plan"):
MODULE.export_plan(
FakeAFCClient({row["remote_path"]: b"12345"}),
[row],
destination,
manifest,
)
self.assertEqual(manifest.read_bytes(), collided_bytes[0])
with self.assertRaisesRegex(RuntimeError, "truncated final event"):
MODULE.load_manifest(manifest)
def test_fault_during_initial_manifest_write_installs_no_manifest(self):
with tempfile.TemporaryDirectory() as directory:
root = Path(directory).resolve()
manifest = root / "manifest.tsv"
row = {
**sample_plan()[0],
"partial_name": "",
"actual_size": "0",
"sha256": "",
"status": "pending",
}
real_write = os.write
def fail_before_first_byte(descriptor, payload):
raise OSError(28, "injected no space")
with mock.patch.object(MODULE.os, "write", side_effect=fail_before_first_byte):
with self.assertRaisesRegex(OSError, "injected no space"):
MODULE.write_manifest_atomic(manifest, [row], expected_plan=[row])
self.assertFalse(manifest.exists())
def fail_after_partial_header(descriptor, payload):
real_write(descriptor, payload[:7])
raise OSError(28, "injected partial header")
with mock.patch.object(MODULE.os, "write", side_effect=fail_after_partial_header):
with self.assertRaisesRegex(OSError, "injected partial header"):
MODULE.write_manifest_atomic(manifest, [row], expected_plan=[row])
self.assertFalse(manifest.exists())
self.assertEqual(list(root.glob("*.partial")), [])
MODULE.write_manifest_atomic(manifest, [row], expected_plan=[row])
self.assertEqual(MODULE.load_manifest(manifest)[row["filename"]]["status"], "pending")
def test_torn_trailing_manifest_event_is_recovered_under_export_lock(self):
with tempfile.TemporaryDirectory() as directory:
root = Path(directory).resolve()
destination = root / "music"
manifest = root / "manifest.tsv"
row = sample_plan()[0]
row["title"] = "Line one\nLine two"
MODULE.export_plan(
FakeAFCClient({row["remote_path"]: b"12345"}),
[row],
destination,
manifest,
)
with manifest.open("ab") as output:
output.write(b"torn-final-event")
with self.assertRaisesRegex(RuntimeError, "truncated final event"):
MODULE.load_manifest(manifest)
summary = MODULE.export_plan(FakeAFCClient({}), [row], destination, manifest)
verification = MODULE.verify_export(
manifest, destination, "Electro", probe_audio=False
)
self.assertEqual(summary["resumed_tracks"], 1)
self.assertEqual(verification["hash_errors"], [])
self.assertTrue(manifest.read_bytes().endswith(b"\n"))
self.assertNotIn(b"Line one\nLine two", manifest.read_bytes())
def test_symlink_destination_and_unexpected_directory_are_rejected(self):
with tempfile.TemporaryDirectory() as directory:
root = Path(directory).resolve()
real_destination = root / "real"
real_destination.mkdir()
linked_destination = root / "linked"
linked_destination.symlink_to(real_destination, target_is_directory=True)
with self.assertRaisesRegex(RuntimeError, "symlink component"):
MODULE.inspect_existing_destination(linked_destination, sample_plan(), {})
(real_destination / "unexpected-directory").mkdir()
with self.assertRaisesRegex(RuntimeError, "entries outside"):
MODULE.inspect_existing_destination(real_destination, sample_plan(), {})
def test_media_paths_are_confined_to_approved_roots(self):
self.assertEqual(
MODULE.make_safe_remote_media_path("iTunes_Control/Music/F00", "ABCD.m4a"),
"/iTunes_Control/Music/F00/ABCD.m4a",
)
with self.assertRaisesRegex(RuntimeError, "unsafe media database path"):
MODULE.make_safe_remote_media_path("iTunes_Control/Music/F00", "../secret")
with self.assertRaisesRegex(RuntimeError, "outside approved roots"):
MODULE.make_safe_remote_media_path("Safari", "history.db")
def test_protocol_dispatchers_block_unknown_operations(self):
with self.assertRaisesRegex(RuntimeError, "blocked AFC operation"):
MODULE.AFCClient(None).request(0xFFFF)
writable_open = struct.pack("<Q", 3) + b"/iTunes_Control/Music/file\0"
with self.assertRaisesRegex(RuntimeError, "non-read-only AFC file-open"):
MODULE.AFCClient(None).request(MODULE.AFC_OP_FILE_OPEN, writable_open)
with self.assertRaisesRegex(RuntimeError, "blocked lockdownd request"):
MODULE.lockdown_request(None, "Pair")
with self.assertRaisesRegex(RuntimeError, "blocked non-read-only usbmuxd"):
MODULE.usbmux_request({"MessageType": "SavePairRecord"})
def test_lockdown_session_cannot_downgrade_paired_tls(self):
class FakeSocket:
def __init__(self):
self.closed = False
def close(self):
self.closed = True
fake_socket = FakeSocket()
with (
mock.patch.object(MODULE, "usbmux_connect", return_value=fake_socket),
mock.patch.object(
MODULE,
"lockdown_request",
return_value={"EnableSessionSSL": False, "SessionID": "session"},
),
mock.patch.object(MODULE, "make_ssl_context") as make_ssl_context,
):
with self.assertRaisesRegex(RuntimeError, "required paired TLS"):
MODULE.start_lockdown_session({"DeviceID": 1}, {"HostID": "h", "SystemBUID": "s"})
self.assertTrue(fake_socket.closed)
make_ssl_context.assert_not_called()
def test_hot_rollback_journal_is_rejected(self):
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
source = root / "source.sqlitedb"
create_media_database(source)
client = FakeAFCClient(
{
MODULE.MEDIA_DATABASE_REMOTE: source.read_bytes(),
MODULE.MEDIA_DATABASE_REMOTE + "-journal": b"hot rollback journal",
}
)
snapshot = MODULE.capture_database_set(client, root / "snapshot-work")
self.assertIsNone(snapshot)
def test_invalid_nested_manifest_creates_no_destination_or_lock(self):
with tempfile.TemporaryDirectory() as directory:
root = Path(directory).resolve()
destination = root / "music"
manifest = destination / "manifest.tsv"
with self.assertRaisesRegex(RuntimeError, "manifest outside"):
MODULE.export_plan(FakeAFCClient({}), [sample_plan()[0]], destination, manifest)
self.assertFalse(destination.exists())
self.assertFalse(Path(str(manifest) + ".lock").exists())
if __name__ == "__main__":
unittest.main()