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)

tests/ahriman/application/test_ahriman.py

@@ -258,13 +258,14 @@ def test_subparsers_patch_add_architecture(parser: argparse.ArgumentParser) -> N
 
 def test_subparsers_patch_list(parser: argparse.ArgumentParser) -> None:
     """
-    patch-list command must imply action, architecture list, lock and report
+    patch-list command must imply action, architecture list, lock, report and unsafe
     """
     args = parser.parse_args(["patch-list", "ahriman"])
     assert args.action == Action.List
     assert args.architecture == [""]
     assert args.lock is None
     assert not args.report
+    assert args.unsafe
 
 
 def test_subparsers_patch_list_architecture(parser: argparse.ArgumentParser) -> None:
@@ -558,12 +559,13 @@ def test_subparsers_repo_sync_architecture(parser: argparse.ArgumentParser) -> N
 
 def test_subparsers_repo_tree(parser: argparse.ArgumentParser) -> None:
     """
-    repo-tree command must imply lock, report and quiet
+    repo-tree command must imply lock, report, quiet and unsafe
     """
     args = parser.parse_args(["repo-tree"])
     assert args.lock is None
     assert not args.report
     assert args.quiet
+    assert args.unsafe
 
 
 def test_subparsers_repo_tree_architecture(parser: argparse.ArgumentParser) -> None:
@@ -619,7 +621,7 @@ def test_subparsers_shell(parser: argparse.ArgumentParser) -> None:
 
 def test_subparsers_user_add(parser: argparse.ArgumentParser) -> None:
     """
-    user-add command must imply action, architecture, lock, report, quiet and unsafe
+    user-add command must imply action, architecture, lock, report and quiet
     """
     args = parser.parse_args(["user-add", "username"])
     assert args.action == Action.Update
@@ -627,7 +629,6 @@ def test_subparsers_user_add(parser: argparse.ArgumentParser) -> None:
     assert args.lock is None
     assert not args.report
     assert args.quiet
-    assert args.unsafe
 
 
 def test_subparsers_user_add_architecture(parser: argparse.ArgumentParser) -> None:
@@ -680,7 +681,7 @@ def test_subparsers_user_list_option_role(parser: argparse.ArgumentParser) -> No
 
 def test_subparsers_user_remove(parser: argparse.ArgumentParser) -> None:
     """
-    user-remove command must imply action, architecture, lock, report, password, quiet, role and unsafe
+    user-remove command must imply action, architecture, lock, report, password and quiet
     """
     args = parser.parse_args(["user-remove", "username"])
     assert args.action == Action.Remove
@@ -689,7 +690,6 @@ def test_subparsers_user_remove(parser: argparse.ArgumentParser) -> None:
     assert not args.report
     assert args.password is not None
     assert args.quiet
-    assert args.unsafe
 
 
 def test_subparsers_user_remove_architecture(parser: argparse.ArgumentParser) -> None:

tests/ahriman/core/status/test_web_client.py

@@ -123,6 +123,9 @@ def test_package_url(web_client: WebClient, package_ahriman: Package) -> None:
     """
     must generate package status url correctly
     """
+    assert web_client._package_url("").startswith(web_client.address)
+    assert web_client._package_url("").endswith(f"/api/v1/packages")
+
     assert web_client._package_url(package_ahriman.base).startswith(web_client.address)
     assert web_client._package_url(package_ahriman.base).endswith(f"/api/v1/packages/{package_ahriman.base}")
 

tests/ahriman/web/test_web.py

@@ -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
     )