selfprivacy-rest-api/tests/test_graphql/test_ssh.py

512 lines
16 KiB
Python
Raw Normal View History

# pylint: disable=redefined-outer-name
# pylint: disable=unused-argument
import pytest
from selfprivacy_api.graphql.mutations.system_mutations import SystemMutations
from selfprivacy_api.graphql.queries.system import System
from tests.common import read_json, generate_system_query
2023-12-15 10:43:07 +00:00
from tests.test_graphql.common import assert_empty, assert_ok, get_data
class ProcessMock:
"""Mock subprocess.Popen"""
def __init__(self, args, **kwargs):
self.args = args
self.kwargs = kwargs
2023-12-15 11:22:20 +00:00
def communicate(self): # pylint: disable=no-method-argument
return (b"NEW_HASHED", None)
returncode = 0
@pytest.fixture
def mock_subprocess_popen(mocker):
mock = mocker.patch("subprocess.Popen", autospec=True, return_value=ProcessMock)
return mock
@pytest.fixture
def some_users(mocker, datadir):
mocker.patch("selfprivacy_api.utils.USERDATA_FILE", new=datadir / "some_users.json")
assert read_json(datadir / "some_users.json")["users"] == [
{
"username": "user1",
"hashedPassword": "HASHED_PASSWORD_1",
"sshKeys": ["ssh-rsa KEY user1@pc"],
},
{"username": "user2", "hashedPassword": "HASHED_PASSWORD_2", "sshKeys": []},
{"username": "user3", "hashedPassword": "HASHED_PASSWORD_3"},
]
return datadir
# TESTS ########################################################
API_CREATE_SSH_KEY_MUTATION = """
mutation addSshKey($sshInput: SshMutationInput!) {
users {
addSshKey(sshInput: $sshInput) {
success
message
code
user {
username
sshKeys
}
}
}
}
"""
API_SET_SSH_SETTINGS = """
2023-12-15 10:43:07 +00:00
mutation enableSsh($settings: SSHSettingsInput!) {
system {
2023-12-15 10:43:07 +00:00
changeSshSettings(settings: $settings) {
success
message
code
enable
passwordAuthentication
}
}
}
"""
API_SSH_SETTINGS_QUERY = """
settings {
ssh {
enable
passwordAuthentication
}
}
"""
def api_ssh_settings_raw(client):
return client.post(
"/graphql",
json={"query": generate_system_query([API_SSH_SETTINGS_QUERY])},
)
def api_ssh_settings(authorized_client):
response = api_ssh_settings_raw(authorized_client)
data = get_data(response)
result = data["system"]["settings"]["ssh"]
assert result is not None
return result
2023-12-15 10:43:07 +00:00
def api_set_ssh_settings(authorized_client, enable: bool, password_auth: bool):
response = authorized_client.post(
"/graphql",
json={
"query": API_SET_SSH_SETTINGS,
"variables": {
"settings": {
"enable": enable,
"passwordAuthentication": password_auth,
},
},
},
)
data = get_data(response)
result = data["system"]["changeSshSettings"]
assert result is not None
return result
def test_graphql_ssh_query(authorized_client, some_users):
settings = api_ssh_settings(authorized_client)
assert settings["enable"] is True
assert settings["passwordAuthentication"] is True
def test_graphql_get_ssh_settings_unauthorized(client, some_users):
response = api_ssh_settings_raw(client)
assert_empty(response)
def test_graphql_change_ssh_settings_unauthorized(client, some_users):
response = client.post(
"/graphql",
json={
"query": API_SET_SSH_SETTINGS,
"variables": {
"sshInput": {
"enable": True,
"passwordAuthentication": True,
},
},
},
)
assert_empty(response)
2023-12-15 11:02:31 +00:00
def assert_includes(smaller_dict: dict, bigger_dict: dict):
for item in smaller_dict.items():
assert item in bigger_dict.items()
2023-12-15 10:51:47 +00:00
def test_graphql_disable_enable_ssh(
authorized_client, some_users, mock_subprocess_popen
):
# Off
2023-12-15 10:43:07 +00:00
output = api_set_ssh_settings(authorized_client, enable=False, password_auth=False)
assert_ok(output)
2023-12-15 11:22:20 +00:00
assert output["enable"] is False
assert output["passwordAuthentication"] is False
2023-12-15 11:02:31 +00:00
assert_includes(api_ssh_settings(authorized_client), output)
2023-12-15 10:43:07 +00:00
# On
2023-12-15 10:51:47 +00:00
output = api_set_ssh_settings(authorized_client, enable=True, password_auth=True)
assert_ok(output)
2023-12-15 11:22:20 +00:00
assert output["enable"] is True
assert output["passwordAuthentication"] is True
2023-12-15 11:02:31 +00:00
assert_includes(api_ssh_settings(authorized_client), output)
2023-12-15 10:51:47 +00:00
# Criss-Cross
2023-12-15 10:51:47 +00:00
output = api_set_ssh_settings(authorized_client, enable=True, password_auth=False)
assert_ok(output)
2023-12-15 11:22:20 +00:00
assert output["enable"] is True
assert output["passwordAuthentication"] is False
2023-12-15 11:02:31 +00:00
assert_includes(api_ssh_settings(authorized_client), output)
2023-12-15 10:51:47 +00:00
output = api_set_ssh_settings(authorized_client, enable=False, password_auth=True)
assert_ok(output)
2023-12-15 11:22:20 +00:00
assert output["enable"] is False
assert output["passwordAuthentication"] is True
2023-12-15 11:02:31 +00:00
assert_includes(api_ssh_settings(authorized_client), output)
2023-12-15 10:51:47 +00:00
2023-12-15 10:43:07 +00:00
2023-12-15 11:22:20 +00:00
def test_graphql_disable_twice(authorized_client, some_users):
output = api_set_ssh_settings(authorized_client, enable=False, password_auth=False)
assert_ok(output)
2023-12-15 11:22:20 +00:00
assert output["enable"] is False
assert output["passwordAuthentication"] is False
output = api_set_ssh_settings(authorized_client, enable=False, password_auth=False)
assert_ok(output)
2023-12-15 11:22:20 +00:00
assert output["enable"] is False
assert output["passwordAuthentication"] is False
2023-12-15 11:22:20 +00:00
def test_graphql_enable_twice(authorized_client, some_users):
output = api_set_ssh_settings(authorized_client, enable=True, password_auth=True)
assert_ok(output)
2023-12-15 11:22:20 +00:00
assert output["enable"] is True
assert output["passwordAuthentication"] is True
assert_includes(api_ssh_settings(authorized_client), output)
output = api_set_ssh_settings(authorized_client, enable=True, password_auth=True)
assert_ok(output)
2023-12-15 11:22:20 +00:00
assert output["enable"] is True
assert output["passwordAuthentication"] is True
assert_includes(api_ssh_settings(authorized_client), output)
def test_graphql_add_ssh_key_unauthorized(client, some_users, mock_subprocess_popen):
response = client.post(
"/graphql",
json={
"query": API_CREATE_SSH_KEY_MUTATION,
"variables": {
"sshInput": {
"username": "user1",
"sshKey": "ssh-rsa KEY test_key@pc",
},
},
},
)
assert_empty(response)
def test_graphql_add_ssh_key(authorized_client, some_users, mock_subprocess_popen):
response = authorized_client.post(
"/graphql",
json={
"query": API_CREATE_SSH_KEY_MUTATION,
"variables": {
"sshInput": {
"username": "user1",
"sshKey": "ssh-rsa KEY test_key@pc",
},
},
},
)
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["users"]["addSshKey"]["code"] == 201
assert response.json()["data"]["users"]["addSshKey"]["message"] is not None
assert response.json()["data"]["users"]["addSshKey"]["success"] is True
assert response.json()["data"]["users"]["addSshKey"]["user"]["username"] == "user1"
assert response.json()["data"]["users"]["addSshKey"]["user"]["sshKeys"] == [
"ssh-rsa KEY user1@pc",
"ssh-rsa KEY test_key@pc",
]
def test_graphql_add_root_ssh_key(authorized_client, some_users, mock_subprocess_popen):
response = authorized_client.post(
"/graphql",
json={
"query": API_CREATE_SSH_KEY_MUTATION,
"variables": {
"sshInput": {
"username": "root",
"sshKey": "ssh-rsa KEY test_key@pc",
},
},
},
)
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["users"]["addSshKey"]["code"] == 201
assert response.json()["data"]["users"]["addSshKey"]["message"] is not None
assert response.json()["data"]["users"]["addSshKey"]["success"] is True
assert response.json()["data"]["users"]["addSshKey"]["user"]["username"] == "root"
assert response.json()["data"]["users"]["addSshKey"]["user"]["sshKeys"] == [
"ssh-ed25519 KEY test@pc",
"ssh-rsa KEY test_key@pc",
]
def test_graphql_add_main_ssh_key(authorized_client, some_users, mock_subprocess_popen):
response = authorized_client.post(
"/graphql",
json={
"query": API_CREATE_SSH_KEY_MUTATION,
"variables": {
"sshInput": {
"username": "tester",
"sshKey": "ssh-rsa KEY test_key@pc",
},
},
},
)
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["users"]["addSshKey"]["code"] == 201
assert response.json()["data"]["users"]["addSshKey"]["message"] is not None
assert response.json()["data"]["users"]["addSshKey"]["success"] is True
assert response.json()["data"]["users"]["addSshKey"]["user"]["username"] == "tester"
assert response.json()["data"]["users"]["addSshKey"]["user"]["sshKeys"] == [
"ssh-rsa KEY test@pc",
"ssh-rsa KEY test_key@pc",
]
def test_graphql_add_bad_ssh_key(authorized_client, some_users, mock_subprocess_popen):
response = authorized_client.post(
"/graphql",
json={
"query": API_CREATE_SSH_KEY_MUTATION,
"variables": {
"sshInput": {
"username": "user1",
"sshKey": "trust me, this is the ssh key",
},
},
},
)
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["users"]["addSshKey"]["code"] == 400
assert response.json()["data"]["users"]["addSshKey"]["message"] is not None
assert response.json()["data"]["users"]["addSshKey"]["success"] is False
def test_graphql_add_ssh_key_nonexistent_user(
authorized_client, some_users, mock_subprocess_popen
):
response = authorized_client.post(
"/graphql",
json={
"query": API_CREATE_SSH_KEY_MUTATION,
"variables": {
"sshInput": {
"username": "user666",
"sshKey": "ssh-rsa KEY test_key@pc",
},
},
},
)
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["users"]["addSshKey"]["code"] == 404
assert response.json()["data"]["users"]["addSshKey"]["message"] is not None
assert response.json()["data"]["users"]["addSshKey"]["success"] is False
API_REMOVE_SSH_KEY_MUTATION = """
mutation removeSshKey($sshInput: SshMutationInput!) {
users {
removeSshKey(sshInput: $sshInput) {
success
message
code
user {
username
sshKeys
}
}
}
}
"""
def test_graphql_remove_ssh_key_unauthorized(client, some_users, mock_subprocess_popen):
response = client.post(
"/graphql",
json={
"query": API_REMOVE_SSH_KEY_MUTATION,
"variables": {
"sshInput": {
"username": "user1",
"sshKey": "ssh-rsa KEY test_key@pc",
},
},
},
)
assert_empty(response)
def test_graphql_remove_ssh_key(authorized_client, some_users, mock_subprocess_popen):
response = authorized_client.post(
"/graphql",
json={
"query": API_REMOVE_SSH_KEY_MUTATION,
"variables": {
"sshInput": {
"username": "user1",
"sshKey": "ssh-rsa KEY user1@pc",
},
},
},
)
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["users"]["removeSshKey"]["code"] == 200
assert response.json()["data"]["users"]["removeSshKey"]["message"] is not None
assert response.json()["data"]["users"]["removeSshKey"]["success"] is True
assert (
response.json()["data"]["users"]["removeSshKey"]["user"]["username"] == "user1"
)
assert response.json()["data"]["users"]["removeSshKey"]["user"]["sshKeys"] == []
def test_graphql_remove_root_ssh_key(
authorized_client, some_users, mock_subprocess_popen
):
response = authorized_client.post(
"/graphql",
json={
"query": API_REMOVE_SSH_KEY_MUTATION,
"variables": {
"sshInput": {
"username": "root",
"sshKey": "ssh-ed25519 KEY test@pc",
},
},
},
)
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["users"]["removeSshKey"]["code"] == 200
assert response.json()["data"]["users"]["removeSshKey"]["message"] is not None
assert response.json()["data"]["users"]["removeSshKey"]["success"] is True
assert (
response.json()["data"]["users"]["removeSshKey"]["user"]["username"] == "root"
)
assert response.json()["data"]["users"]["removeSshKey"]["user"]["sshKeys"] == []
def test_graphql_remove_main_ssh_key(
authorized_client, some_users, mock_subprocess_popen
):
response = authorized_client.post(
"/graphql",
json={
"query": API_REMOVE_SSH_KEY_MUTATION,
"variables": {
"sshInput": {
"username": "tester",
"sshKey": "ssh-rsa KEY test@pc",
},
},
},
)
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["users"]["removeSshKey"]["code"] == 200
assert response.json()["data"]["users"]["removeSshKey"]["message"] is not None
assert response.json()["data"]["users"]["removeSshKey"]["success"] is True
assert (
response.json()["data"]["users"]["removeSshKey"]["user"]["username"] == "tester"
)
assert response.json()["data"]["users"]["removeSshKey"]["user"]["sshKeys"] == []
def test_graphql_remove_nonexistent_ssh_key(
authorized_client, some_users, mock_subprocess_popen
):
response = authorized_client.post(
"/graphql",
json={
"query": API_REMOVE_SSH_KEY_MUTATION,
"variables": {
"sshInput": {
"username": "user1",
"sshKey": "ssh-rsa KEY test_key@pc",
},
},
},
)
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["users"]["removeSshKey"]["code"] == 404
assert response.json()["data"]["users"]["removeSshKey"]["message"] is not None
assert response.json()["data"]["users"]["removeSshKey"]["success"] is False
def test_graphql_remove_ssh_key_nonexistent_user(
authorized_client, some_users, mock_subprocess_popen
):
response = authorized_client.post(
"/graphql",
json={
"query": API_REMOVE_SSH_KEY_MUTATION,
"variables": {
"sshInput": {
"username": "user666",
"sshKey": "ssh-rsa KEY test_key@pc",
},
},
},
)
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["users"]["removeSshKey"]["code"] == 404
assert response.json()["data"]["users"]["removeSshKey"]["message"] is not None
assert response.json()["data"]["users"]["removeSshKey"]["success"] is False