107 lines
5.2 KiB
Python
107 lines
5.2 KiB
Python
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
import subprocess
|
|
|
|
import pytest
|
|
|
|
from codexpi.database import Database
|
|
from codexpi.ssh_service import REMOTE_SHELL_SCRIPT, SSHError, _parse_shell_response, _shell_argument, create_project, create_server_project, list_servers, request, save_server, server_request, validate_connection
|
|
|
|
|
|
def test_ssh_connection_fields_reject_command_injection():
|
|
assert validate_connection("192.168.1.172", 22, "ubuntu") == ("192.168.1.172", 22, "ubuntu")
|
|
with pytest.raises(SSHError):
|
|
validate_connection("server; reboot", 22, "ubuntu")
|
|
with pytest.raises(SSHError):
|
|
validate_connection("server.local", 22, "root$(id)")
|
|
|
|
|
|
def test_shell_fallback_browses_without_python():
|
|
arguments = ["browse", _shell_argument("/home"), _shell_argument(""), _shell_argument(""), _shell_argument("")]
|
|
result = subprocess.run(["sh", "-c", REMOTE_SHELL_SCRIPT, "--", *arguments], capture_output=True, timeout=10)
|
|
parsed = _parse_shell_response(result.stdout)
|
|
|
|
assert result.returncode == 0
|
|
assert parsed["path"] == "/home" and parsed["parent"] == "/"
|
|
assert all(item["path"].startswith("/home/") for item in parsed["directories"])
|
|
|
|
|
|
def test_ssh_project_uses_remote_project_channel(tmp_path: Path, monkeypatch):
|
|
database = Database(tmp_path / "codexpi.db")
|
|
database.initialize()
|
|
|
|
def fake_request(_data_dir, host, port, username, operation, path, **_kwargs):
|
|
assert (host, port, username, operation, path) == ("server.local", 2222, "dev", "browse", "/home/dev/app")
|
|
return {"path": path, "parent": "/home/dev", "selectable": True, "directories": []}
|
|
|
|
monkeypatch.setattr("codexpi.ssh_service.request", fake_request)
|
|
project = create_project(database, tmp_path, "secret", "Application", "server.local", 2222, "dev", "/home/dev/app")
|
|
|
|
assert project["kind"] == "remote"
|
|
machine = database.one("SELECT * FROM remote_machines WHERE project_id=?", (project["id"],))
|
|
assert machine and machine["name"] == "dev@server.local"
|
|
connection = database.one("SELECT * FROM ssh_connections WHERE machine_id=?", (machine["id"],))
|
|
assert connection and connection["folder"] == "/home/dev/app" and connection["port"] == 2222
|
|
assert (tmp_path / "remote-workspaces" / project["id"]).is_dir()
|
|
|
|
|
|
def test_password_is_passed_by_file_descriptor_not_command_line(tmp_path: Path, monkeypatch):
|
|
received = {}
|
|
monkeypatch.setattr("codexpi.ssh_service.ensure_key", lambda _data_dir: {})
|
|
|
|
def fake_run(command, **kwargs):
|
|
received["command"] = command
|
|
received["password"] = __import__("os").read(kwargs["pass_fds"][0], 1024)
|
|
return SimpleNamespace(returncode=0, stdout=b'{"ok":true,"result":{"path":"/home"}}\n', stderr=b"")
|
|
|
|
monkeypatch.setattr("codexpi.ssh_service.subprocess.run", fake_run)
|
|
result = request(tmp_path, "server.local", 22, "dev", "browse", auth_method="password", password="top-secret")
|
|
|
|
assert result["path"] == "/home"
|
|
assert received["password"] == b"top-secret\n"
|
|
assert "top-secret" not in " ".join(received["command"])
|
|
assert received["command"][:3] == ["sshpass", "-d", received["command"][2]]
|
|
|
|
|
|
def test_ssh_project_encrypts_password(tmp_path: Path, monkeypatch):
|
|
database = Database(tmp_path / "codexpi.db")
|
|
database.initialize()
|
|
monkeypatch.setattr(
|
|
"codexpi.ssh_service.request",
|
|
lambda *_args, **_kwargs: {"path": "/home/dev/app", "parent": "/home/dev", "selectable": True, "directories": []},
|
|
)
|
|
|
|
project = create_project(
|
|
database, tmp_path, "application-secret", "Application", "server.local", 22, "dev",
|
|
"/home/dev/app", "password", "top-secret",
|
|
)
|
|
connection = database.one(
|
|
"SELECT sc.* FROM ssh_connections sc JOIN remote_machines rm ON rm.id=sc.machine_id WHERE rm.project_id=?",
|
|
(project["id"],),
|
|
)
|
|
|
|
assert connection and connection["auth_method"] == "password"
|
|
assert connection["password_encrypted"] and "top-secret" not in connection["password_encrypted"]
|
|
|
|
|
|
def test_saved_server_is_sanitized_reusable_and_encrypted(tmp_path: Path, monkeypatch):
|
|
database = Database(tmp_path / "codexpi.db")
|
|
database.initialize()
|
|
calls = []
|
|
|
|
def fake_request(_data_dir, host, port, username, operation, path="/home", **kwargs):
|
|
calls.append((host, port, username, operation, path, kwargs.get("password")))
|
|
return {"path": path, "parent": "/home", "selectable": path != "/home", "directories": []}
|
|
|
|
monkeypatch.setattr("codexpi.ssh_service.request", fake_request)
|
|
server = save_server(database, tmp_path, "application-secret", "Maison", "server.local", 22, "dev", "password", "top-secret")
|
|
|
|
assert server["name"] == "Maison" and "password_encrypted" not in server
|
|
assert list_servers(database)[0]["auth_method"] == "password"
|
|
stored = database.one("SELECT * FROM ssh_servers WHERE id=?", (server["id"],))
|
|
assert stored and "top-secret" not in stored["password_encrypted"]
|
|
server_request(database, tmp_path, "application-secret", server["id"], "browse", "/home/dev")
|
|
project = create_server_project(database, tmp_path, "application-secret", server["id"], "App", "/home/dev/app")
|
|
assert project["kind"] == "remote"
|
|
assert calls[-1][-1] == "top-secret"
|