mirror of
https://github.com/arcan1s/ahriman.git
synced 2025-07-22 18:29:56 +00:00
review unsafe commands access
Some commands were made unsafe in old versions, but nowadays they can be run without having special privileges. There was also a bug in which status commands were not available if you are not ahriman user and unix socket is used. It has been fixed by switching to manual socket creation (see also https://github.com/aio-libs/aiohttp/issues/4155)
This commit is contained in:
@ -1,12 +1,62 @@
|
||||
import pytest
|
||||
import socket
|
||||
|
||||
from aiohttp import web
|
||||
from pytest_mock import MockerFixture
|
||||
from unittest.mock import call as MockCall
|
||||
|
||||
from ahriman.core.exceptions import InitializeError
|
||||
from ahriman.core.log.filtered_access_logger import FilteredAccessLogger
|
||||
from ahriman.core.status.watcher import Watcher
|
||||
from ahriman.web.web import on_shutdown, on_startup, run_server
|
||||
from ahriman.web.web import create_socket, on_shutdown, on_startup, run_server
|
||||
|
||||
|
||||
async def test_create_socket(application: web.Application, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must create socket
|
||||
"""
|
||||
path = "/run/ahriman.sock"
|
||||
application["configuration"].set_option("web", "unix_socket", str(path))
|
||||
current_on_shutdown = len(application.on_shutdown)
|
||||
|
||||
bind_mock = mocker.patch("socket.socket.bind")
|
||||
chmod_mock = mocker.patch("pathlib.Path.chmod")
|
||||
unlink_mock = mocker.patch("pathlib.Path.unlink")
|
||||
|
||||
sock = create_socket(application["configuration"], application)
|
||||
assert sock.family == socket.AF_UNIX
|
||||
assert sock.type == socket.SOCK_STREAM
|
||||
bind_mock.assert_called_once_with(str(path))
|
||||
chmod_mock.assert_called_once_with(0o666)
|
||||
assert len(application.on_shutdown) == current_on_shutdown + 1
|
||||
|
||||
# provoke socket removal
|
||||
await application.on_shutdown[-1](application)
|
||||
unlink_mock.assert_has_calls([MockCall(missing_ok=True), MockCall(missing_ok=True)])
|
||||
|
||||
|
||||
def test_create_socket_empty(application: web.Application) -> None:
|
||||
"""
|
||||
must skip socket creation if not set by configuration
|
||||
"""
|
||||
assert create_socket(application["configuration"], application) is None
|
||||
|
||||
|
||||
def test_create_socket_safe(application: web.Application, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must create socket with default permission set
|
||||
"""
|
||||
path = "/run/ahriman.sock"
|
||||
application["configuration"].set_option("web", "unix_socket", str(path))
|
||||
application["configuration"].set_option("web", "unix_socket_unsafe", "no")
|
||||
|
||||
mocker.patch("socket.socket.bind")
|
||||
mocker.patch("pathlib.Path.unlink")
|
||||
chmod_mock = mocker.patch("pathlib.Path.chmod")
|
||||
|
||||
sock = create_socket(application["configuration"], application)
|
||||
assert sock is not None
|
||||
chmod_mock.assert_not_called()
|
||||
|
||||
|
||||
async def test_on_shutdown(application: web.Application, mocker: MockerFixture) -> None:
|
||||
@ -50,7 +100,7 @@ def test_run(application: web.Application, mocker: MockerFixture) -> None:
|
||||
|
||||
run_server(application)
|
||||
run_application_mock.assert_called_once_with(
|
||||
application, host="127.0.0.1", port=port, path=None, handle_signals=False,
|
||||
application, host="127.0.0.1", port=port, sock=None, handle_signals=False,
|
||||
access_log=pytest.helpers.anyvar(int), access_log_class=FilteredAccessLogger
|
||||
)
|
||||
|
||||
@ -65,7 +115,7 @@ def test_run_with_auth(application_with_auth: web.Application, mocker: MockerFix
|
||||
|
||||
run_server(application_with_auth)
|
||||
run_application_mock.assert_called_once_with(
|
||||
application_with_auth, host="127.0.0.1", port=port, path=None, handle_signals=False,
|
||||
application_with_auth, host="127.0.0.1", port=port, sock=None, handle_signals=False,
|
||||
access_log=pytest.helpers.anyvar(int), access_log_class=FilteredAccessLogger
|
||||
)
|
||||
|
||||
@ -80,7 +130,7 @@ def test_run_with_debug(application_with_debug: web.Application, mocker: MockerF
|
||||
|
||||
run_server(application_with_debug)
|
||||
run_application_mock.assert_called_once_with(
|
||||
application_with_debug, host="127.0.0.1", port=port, path=None, handle_signals=False,
|
||||
application_with_debug, host="127.0.0.1", port=port, sock=None, handle_signals=False,
|
||||
access_log=pytest.helpers.anyvar(int), access_log_class=FilteredAccessLogger
|
||||
)
|
||||
|
||||
@ -90,13 +140,13 @@ def test_run_with_socket(application: web.Application, mocker: MockerFixture) ->
|
||||
must run application
|
||||
"""
|
||||
port = 8080
|
||||
socket = "/run/ahriman.sock"
|
||||
application["configuration"].set_option("web", "port", str(port))
|
||||
application["configuration"].set_option("web", "unix_socket", socket)
|
||||
socket_mock = mocker.patch("ahriman.web.web.create_socket", return_value=42)
|
||||
run_application_mock = mocker.patch("aiohttp.web.run_app")
|
||||
|
||||
run_server(application)
|
||||
socket_mock.assert_called_once_with(application["configuration"], application)
|
||||
run_application_mock.assert_called_once_with(
|
||||
application, host="127.0.0.1", port=port, path=socket, handle_signals=False,
|
||||
application, host="127.0.0.1", port=port, sock=42, handle_signals=False,
|
||||
access_log=pytest.helpers.anyvar(int), access_log_class=FilteredAccessLogger
|
||||
)
|
||||
|
Reference in New Issue
Block a user