from __future__ import annotations import importlib.util import threading import time from pathlib import Path import pytest from codexpi.database import Database, utcnow from codexpi.remote_service import RemoteError, RemoteService, enqueue_remote_job from codexpi.remote_ipc import RemoteIPCManager, remote_channel_request def load_bridge(): path = Path(__file__).parents[1] / "scripts" / "codexpi_bridge.py" spec = importlib.util.spec_from_file_location("codexpi_bridge_test", path) assert spec and spec.loader module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module def test_remote_pairing_is_one_time_and_revocable(logged_in, monkeypatch): client, app, projects, _, headers = logged_in root = projects / "remote-project" root.mkdir() project = client.post("/api/projects", headers=headers, json={"name": "Remote", "path": str(root)}).json() created = client.post(f"/api/projects/{project['id']}/remote-pairings", headers=headers) assert created.status_code == 201 pairing = created.json() stored = app.state.db.one("SELECT code_hash FROM remote_pairings WHERE id=?", (pairing["pairing_id"],)) assert stored and stored["code_hash"] != pairing["code"] body = { "pairing_id": pairing["pairing_id"], "code": pairing["code"], "name": "PC bureau", "folder_name": "Projet", "platform": "Windows", } connected = client.post("/api/remote/agent/pair", json=body) assert connected.status_code == 201 identity = connected.json() assert "token" in identity assert client.post("/api/remote/agent/pair", json=body).status_code == 401 authorization = {"Authorization": f"Bearer {identity['token']}"} idle = client.get(f"/api/remote/agent/jobs?machine_id={identity['machine_id']}", headers=authorization) assert idle.status_code == 204 remote_project = app.state.db.one("SELECT * FROM projects WHERE id=?", (identity["project_id"],)) assert remote_project and remote_project["kind"] == "remote" and remote_project["name"] == "Projet" assert Path(remote_project["path"]).is_dir() assert (Path(remote_project["path"]) / ".codexpi-remote" / "requests").is_dir() machines = client.get(f"/api/projects/{identity['project_id']}/remote-machines").json() assert len(machines) == 1 and machines[0]["online"] is True assert "token" not in machines[0] all_machines = client.get("/api/remote/machines").json() assert all_machines[0]["project_id"] == identity["project_id"] assert all_machines[0]["project_name"] == "Projet" conversation = client.post( f"/api/projects/{identity['project_id']}/conversations", headers=headers, json={"title": "Distant"} ) assert conversation.status_code == 201 app.state.db.execute( "UPDATE remote_machines SET last_seen=? WHERE id=?", ("2000-01-01T00:00:00+00:00", identity["machine_id"]), ) assert client.get(f"/api/conversations/{conversation.json()['id']}/messages").status_code == 200 app.state.codex.codex_bin = __import__('sys').executable monkeypatch.setattr(app.state.codex, "_spawn", lambda *_: 12345) assert client.post( f"/api/conversations/{conversation.json()['id']}/queue", headers=headers, json={"content": "Bonjour"} ).status_code == 202 assert client.post( f"/api/projects/{identity['project_id']}/conversations", headers=headers, json={"title": "Consultation"} ).status_code == 201 app.state.db.execute("UPDATE prompt_queue SET status='done' WHERE conversation_id=?", (conversation.json()["id"],)) app.state.db.execute("UPDATE conversations SET status='completed',worker_pid=NULL WHERE id=?", (conversation.json()["id"],)) revoked = client.delete(f"/api/remote/machines/{identity['machine_id']}", headers=headers) assert revoked.status_code == 200 assert client.post( f"/api/projects/{identity['project_id']}/conversations", headers=headers, json={"title": "Révoquée"} ).status_code == 409 assert client.get(f"/api/remote/agent/jobs?machine_id={identity['machine_id']}", headers=authorization).status_code == 401 revoked_project = next(item for item in client.get("/api/projects").json() if item["id"] == identity["project_id"]) assert revoked_project["name"] == "[revok] Projet" assert revoked_project["remote_revoked"] is True archived = client.put( f"/api/projects/{identity['project_id']}/archive", headers=headers, json={"archived": True}, ) assert archived.status_code == 200 assert all(item["id"] != identity["project_id"] for item in client.get("/api/projects").json()) assert any(item["id"] == identity["project_id"] for item in client.get("/api/projects?archived=true").json()) assert client.get(f"/api/conversations/{conversation.json()['id']}/messages").status_code == 200 restored = client.put( f"/api/projects/{identity['project_id']}/archive", headers=headers, json={"archived": False}, ) assert restored.status_code == 200 assert any(item["id"] == identity["project_id"] for item in client.get("/api/projects").json()) def test_remote_project_cannot_be_archived_before_machine_is_revoked(logged_in): client, app, projects, _, headers = logged_in root = projects / "archive-guard" root.mkdir() source = client.post("/api/projects", headers=headers, json={"name": "Source", "path": str(root)}).json() pairing = client.post(f"/api/projects/{source['id']}/remote-pairings", headers=headers).json() identity = client.post("/api/remote/agent/pair", json={ "pairing_id": pairing["pairing_id"], "code": pairing["code"], "name": "PC actif", "folder_name": "Projet actif", "platform": "Windows", }).json() response = client.put( f"/api/projects/{identity['project_id']}/archive", headers=headers, json={"archived": True}, ) assert response.status_code == 409 assert app.state.db.one("SELECT archived FROM projects WHERE id=?", (identity["project_id"],))["archived"] == 0 def test_remote_job_round_trip_and_machine_scope(tmp_path): db = Database(tmp_path / "remote.sqlite") db.initialize() db.execute("INSERT INTO projects(id,name,path,created_at) VALUES(?,?,?,?)", ("project", "Remote", str(tmp_path), utcnow())) service = RemoteService(db, "test-secret") pairing = service.create_pairing("project") identity = service.pair(pairing["pairing_id"], pairing["code"], "Laptop", "Sources", "Linux") service.poll(identity["machine_id"], identity["token"]) remote_project_id = identity["project_id"] outcome: dict = {} def request_read(): outcome.update(enqueue_remote_job(str(db.path), remote_project_id, "read", "notes.txt", timeout=5)) worker = threading.Thread(target=request_read) worker.start() job = None for _ in range(30): job = service.poll(identity["machine_id"], identity["token"]) if job: break time.sleep(0.05) assert job and job["operation"] == "read" and job["path"] == "notes.txt" service.complete(identity["machine_id"], identity["token"], job["id"], True, {"content": "bonjour", "sha256": "abc"}, "") worker.join(timeout=5) assert not worker.is_alive() assert outcome == {"content": "bonjour", "sha256": "abc"} with pytest.raises(RemoteError, match="hors ligne"): db.execute("UPDATE remote_machines SET last_seen=? WHERE id=?", ("2000-01-01T00:00:00+00:00", identity["machine_id"])) enqueue_remote_job(str(db.path), remote_project_id, "list", timeout=1) def test_remote_operation_allowlist(tmp_path): db = Database(tmp_path / "remote.sqlite") db.initialize() with pytest.raises(RemoteError, match="interdite"): enqueue_remote_job(str(db.path), "project", "shell", payload={"command": "whoami"}) def test_project_scoped_ipc_proxies_remote_jobs_without_exposing_database(tmp_path): db = Database(tmp_path / "remote.sqlite") db.initialize() db.execute("INSERT INTO projects(id,name,path,created_at) VALUES(?,?,?,?)", ("source", "Remote", str(tmp_path), utcnow())) service = RemoteService(db, "test-secret") pairing = service.create_pairing("source") identity = service.pair(pairing["pairing_id"], pairing["code"], "Laptop", "Sources", "Linux") service.poll(identity["machine_id"], identity["token"]) channel_root = tmp_path / "ipc" manager = RemoteIPCManager(db) channel_path = manager.ensure(identity["project_id"], channel_root) try: machines = remote_channel_request(channel_path, {"operation": "machines"}) assert isinstance(machines, list) and machines[0]["name"] == "Laptop" outcome: dict = {} def request_list(): outcome.update(remote_channel_request(channel_path, {"operation": "list", "path": ""})) worker = threading.Thread(target=request_list) worker.start() job = None for _ in range(30): job = service.poll(identity["machine_id"], identity["token"]) if job: break time.sleep(0.05) assert job and job["operation"] == "list" service.complete(identity["machine_id"], identity["token"], job["id"], True, {"entries": []}, "") worker.join(timeout=5) assert outcome == {"entries": []} finally: manager.close() def test_existing_machine_is_migrated_to_its_own_remote_project(tmp_path): db = Database(tmp_path / "legacy.sqlite") db.initialize() stamp = utcnow() db.execute("INSERT INTO projects(id,name,path,created_at) VALUES(?,?,?,?)", ("local", "CodexPi", str(tmp_path / "local"), stamp)) db.execute( "INSERT INTO remote_machines(id,project_id,name,folder_name,platform,token_hash,status,last_seen,created_at) VALUES(?,?,?,?,?,?,?,?,?)", ("machine", "local", "PC", "Dossier client", "Windows", "hash", "online", stamp, stamp), ) db.initialize() machine = db.one("SELECT project_id FROM remote_machines WHERE id='machine'") assert machine and machine["project_id"] != "local" project = db.one("SELECT * FROM projects WHERE id=?", (machine["project_id"],)) assert project and project["kind"] == "remote" and project["name"] == "Dossier client" assert Path(project["path"]).is_dir() def test_bridge_confines_files_and_prevents_stale_overwrite(tmp_path): bridge = load_bridge() root = tmp_path / "shared" root.mkdir() created = bridge.write_file(root, "notes/todo.txt", {"content": "bonjour"}, None) loaded = bridge.read_file(root, "notes/todo.txt") assert loaded["content"] == "bonjour" assert loaded["sha256"] == created["sha256"] bridge.write_file(root, "notes/todo.txt", {"content": ""}, loaded["sha256"]) assert bridge.read_file(root, "notes/todo.txt")["content"] == "" with pytest.raises(ValueError, match="Conflit"): bridge.write_file(root, "notes/todo.txt", {"content": "ancien"}, loaded["sha256"]) with pytest.raises(ValueError, match="invalide"): bridge.safe_path(root, "../secret.txt") def test_bridge_rejects_symlinks_and_binary_files(tmp_path): bridge = load_bridge() root = tmp_path / "shared" root.mkdir() outside = tmp_path / "outside.txt" outside.write_text("secret", encoding="utf-8") (root / "outside-link").symlink_to(outside) (root / "binary.dat").write_bytes(b"hello\x00world") with pytest.raises(ValueError, match="dehors du dossier partagé"): bridge.read_file(root, "outside-link") with pytest.raises(ValueError, match="binaires"): bridge.read_file(root, "binary.dat") def test_lightweight_bridge_executes_argument_arrays_inside_shared_folder(tmp_path): bridge = load_bridge() root = tmp_path / "shared" root.mkdir() result = bridge.run_command(root, {"argv": ["sh", "-c", "pwd; printf done"], "timeout": 5}) assert result["returncode"] == 0 assert str(root) in result["output"] and result["output"].endswith("done") with pytest.raises(ValueError, match="invalide"): bridge.run_command(root, {"argv": "touch escaped"})