import argparse import asyncio import contextlib import errno import getpass import grp import json import os import pwd import shlex import stat import subprocess import sys import unittest.mock from collections import deque from pathlib import Path from typing import AsyncGenerator, Dict, Generator, Iterator, Sequence, Union import pytest import pytest_asyncio from cockpit._vendor.systemd_ctypes import bus from cockpit.bridge import Bridge from cockpit.channel import AsyncChannel, Channel, ChannelRoutingRule from cockpit.channels import CHANNEL_TYPES from cockpit.channels.filesystem import tag_from_path from cockpit.jsonutil import JsonDict, JsonObject, JsonValue, get_bool, get_dict, get_int, get_str, json_merge_patch from cockpit.packages import BridgeConfig from cockpit.peer import ConfiguredPeer from cockpit.superuser import is_valid_superuser_config from .mocktransport import MOCK_HOSTNAME, MockTransport class test_iface(bus.Object): sig = bus.Interface.Signal('s') prop = bus.Interface.Property('s', value='none') @bus.Interface.Method('s') def get_prop(self) -> str: return self.prop @pytest.fixture def bridge() -> Bridge: bridge = Bridge(argparse.Namespace(privileged=False, beipack=False)) bridge.superuser_bridges = list(bridge.superuser_rule.bridges) # type: ignore[attr-defined] return bridge def add_pseudo(bridge: Bridge) -> None: bridge.more_superuser_bridges = [*bridge.superuser_bridges, 'pseudo'] # type: ignore[attr-defined] assert bridge.packages is not None # Add pseudo to the existing set of superuser rules bridge.superuser_rule.set_configs([ *bridge.packages.get_bridge_configs(), BridgeConfig({ 'label': 'pseudo', 'spawn': [ sys.executable, os.path.abspath(f'{__file__}/../pseudo.py'), sys.executable, '-m', 'cockpit.bridge', '--privileged' ], 'environ': [ f'PYTHONPATH={":".join(sys.path)}' ], 'privileged': True }) ]) @pytest_asyncio.fixture async def no_init_transport(bridge: Bridge) -> AsyncGenerator[MockTransport, None]: transport = MockTransport(bridge) try: yield transport finally: await transport.stop() @pytest.fixture def transport(no_init_transport: MockTransport) -> MockTransport: no_init_transport.init() return no_init_transport @pytest.mark.asyncio async def test_echo(transport: MockTransport) -> None: echo = await transport.check_open('echo') transport.send_data(echo, b'foo') await transport.assert_data(echo, b'foo') transport.send_ping(channel=echo) await transport.assert_msg('', command='pong', channel=echo) transport.send_done(echo) await transport.assert_msg('', command='done', channel=echo) await transport.assert_msg('', command='close', channel=echo) @pytest.mark.asyncio async def test_echo_invalid_utf8(transport: MockTransport) -> None: echo = await transport.check_open('echo') # invalid UTF-8 closes the channel with protocol-error transport.send_data(echo, b'\x80\xff\xfe') await transport.assert_msg('', command='close', channel=echo, problem='protocol-error') @pytest.mark.asyncio async def test_host(transport: MockTransport) -> None: # try to open a null channel, explicitly naming our host await transport.check_open('null', host=MOCK_HOSTNAME) # try to open a null channel, no host await transport.check_open('null') # try to open a null channel, a different host which is sure to fail DNS await transport.check_open('null', host='¡invalid!', problem='no-host') # make sure host check happens before superuser # ie: requesting superuser=True on another host should fail because we # can't contact the other host ('no-host'), rather than trying to first # go to our superuser bridge ('access-denied') await transport.check_open('null', host='¡invalid!', superuser=True, problem='no-host') # but make sure superuser is indeed failing as we expect, on our host await transport.check_open('null', host=MOCK_HOSTNAME, superuser=True, problem='access-denied') @pytest.mark.asyncio async def test_dbus_call_internal(bridge: Bridge, transport: MockTransport) -> None: my_object = test_iface() bridge.internal_bus.export('/foo', my_object) assert my_object._dbus_bus == bridge.internal_bus.server assert my_object._dbus_path == '/foo' values, = await transport.check_bus_call('/foo', 'org.freedesktop.DBus.Properties', 'GetAll', ["test.iface"]) assert values == {'Prop': {'t': 's', 'v': 'none'}} result, = await transport.check_bus_call('/foo', 'test.iface', 'GetProp', []) assert result == 'none' @pytest.mark.asyncio async def test_dbus_watch(bridge: Bridge, transport: MockTransport) -> None: my_object = test_iface() bridge.internal_bus.export('/foo', my_object) assert my_object._dbus_bus == bridge.internal_bus.server assert my_object._dbus_path == '/foo' # Add a watch internal = await transport.ensure_internal_bus() transport.send_json(internal, watch={'path': '/foo', 'interface': 'test.iface'}, id='4') meta = await transport.next_msg(internal) assert meta['meta']['test.iface'] == { 'methods': {'GetProp': {'in': [], 'out': ['s']}}, 'properties': {'Prop': {'flags': 'r', 'type': 's'}}, 'signals': {'Sig': {'in': ['s']}} } notify = await transport.next_msg(internal) assert notify['notify']['/foo'] == {'test.iface': {'Prop': 'none'}} reply = await transport.next_msg(internal) assert reply == {'id': '4', 'reply': []} # Change a property my_object.prop = 'xyz' notify = await transport.next_msg(internal) assert 'notify' in notify assert notify['notify']['/foo'] == {'test.iface': {'Prop': 'xyz'}} async def verify_root_bridge_not_running(bridge: Bridge, transport: MockTransport) -> None: assert bridge.superuser_rule.peer is None await transport.assert_bus_props('/superuser', 'cockpit.Superuser', {'Bridges': bridge.more_superuser_bridges, 'Current': 'none'}) # type: ignore[attr-defined] null = await transport.check_open('null', superuser=True, problem='access-denied') assert null not in bridge.open_channels @pytest.mark.asyncio async def verify_root_bridge_running(bridge: Bridge, transport: MockTransport) -> None: await transport.assert_bus_props('/superuser', 'cockpit.Superuser', {'Bridges': bridge.more_superuser_bridges, 'Current': 'pseudo'}) # type: ignore[attr-defined] assert bridge.superuser_rule.peer is not None # try to open dbus on the root bridge root_dbus = await transport.check_open('dbus-json3', bus='internal', superuser=True) # verify that the bridge thinks that it's the root bridge await transport.assert_bus_props('/superuser', 'cockpit.Superuser', {'Bridges': [], 'Current': 'root'}, bus=root_dbus) # close up await transport.check_close(channel=root_dbus) @pytest.mark.asyncio async def test_superuser_dbus(bridge: Bridge, transport: MockTransport) -> None: add_pseudo(bridge) await verify_root_bridge_not_running(bridge, transport) # start the superuser bridge -- no password, so it should work straight away () = await transport.check_bus_call('/superuser', 'cockpit.Superuser', 'Start', ['pseudo']) await verify_root_bridge_running(bridge, transport) # open a channel on the root bridge root_null = await transport.check_open('null', superuser=True) # stop the bridge stop = transport.send_bus_call(transport.internal_bus, '/superuser', 'cockpit.Superuser', 'Stop', []) # that should have implicitly closed the open channel await transport.assert_msg('', command='close', channel=root_null) assert root_null not in bridge.open_channels # The Stop method call is done now await transport.assert_msg(transport.internal_bus, reply=[[]], id=stop) def format_methods(methods: Dict[str, str]): return {name: {'t': 'a{sv}', 'v': {'label': {'t': 's', 'v': label}}} for name, label in methods.items()} @pytest.mark.asyncio async def test_superuser_dbus_pw(bridge: Bridge, transport: MockTransport, monkeypatch) -> None: monkeypatch.setenv('PSEUDO_PASSWORD', 'p4ssw0rd') add_pseudo(bridge) await verify_root_bridge_not_running(bridge, transport) # watch for signals await transport.add_bus_match('/superuser', 'cockpit.Superuser') await transport.watch_bus('/superuser', 'cockpit.Superuser', { 'Bridges': bridge.more_superuser_bridges, # type: ignore[attr-defined] 'Current': 'none', 'Methods': format_methods({'pseudo': 'pseudo'}), }) # start the bridge. with a password this is more complicated start = transport.send_bus_call(transport.internal_bus, '/superuser', 'cockpit.Superuser', 'Start', ['pseudo']) # first, init state await transport.assert_bus_notify('/superuser', 'cockpit.Superuser', {'Current': 'init'}) # then, we'll be asked for a password await transport.assert_bus_signal('/superuser', 'cockpit.Superuser', 'Prompt', ['', 'can haz pw?', '', False, '']) # give it await transport.check_bus_call('/superuser', 'cockpit.Superuser', 'Answer', ['p4ssw0rd']) # and now the bridge should be running await transport.assert_bus_notify('/superuser', 'cockpit.Superuser', {'Current': 'pseudo'}) # Start call is now done await transport.assert_bus_reply(start, []) # double-check await verify_root_bridge_running(bridge, transport) @pytest.mark.asyncio async def test_superuser_dbus_wrong_pw(bridge: Bridge, transport: MockTransport, monkeypatch) -> None: monkeypatch.setenv('PSEUDO_PASSWORD', 'p4ssw0rd') add_pseudo(bridge) await verify_root_bridge_not_running(bridge, transport) # watch for signals await transport.add_bus_match('/superuser', 'cockpit.Superuser') await transport.watch_bus('/superuser', 'cockpit.Superuser', { 'Bridges': bridge.more_superuser_bridges, # type: ignore[attr-defined] 'Current': 'none', 'Methods': format_methods({'pseudo': 'pseudo'}), }) # start the bridge. with a password this is more complicated start = transport.send_bus_call(transport.internal_bus, '/superuser', 'cockpit.Superuser', 'Start', ['pseudo']) # first, init state await transport.assert_bus_notify('/superuser', 'cockpit.Superuser', {'Current': 'init'}) # then, we'll be asked for a password await transport.assert_bus_signal('/superuser', 'cockpit.Superuser', 'Prompt', ['', 'can haz pw?', '', False, '']) # give it await transport.check_bus_call('/superuser', 'cockpit.Superuser', 'Answer', ['p5ssw0rd']) # wrong password # pseudo fails after the first wrong attempt await transport.assert_bus_notify('/superuser', 'cockpit.Superuser', {'Current': 'none'}) # Start call is now done and returned failure await transport.assert_bus_error(start, 'cockpit.Superuser.Error', 'pseudo says: Bad password') # double-check await verify_root_bridge_not_running(bridge, transport) @pytest.mark.asyncio async def test_superuser_init(bridge: Bridge, no_init_transport: MockTransport) -> None: add_pseudo(bridge) no_init_transport.init(superuser={"id": "pseudo"}) transport = no_init_transport # this should work right away without auth await transport.assert_msg('', command='superuser-init-done') await verify_root_bridge_running(bridge, transport) @pytest.mark.asyncio async def test_superuser_init_pw(bridge: Bridge, no_init_transport: MockTransport, monkeypatch) -> None: monkeypatch.setenv('PSEUDO_PASSWORD', 'p4ssw0rd') add_pseudo(bridge) no_init_transport.init(superuser={"id": "pseudo"}) transport = no_init_transport msg = await transport.assert_msg('', command='authorize') transport.send_json('', command='authorize', cookie=msg['cookie'], response='p4ssw0rd') # that should have worked await transport.assert_msg('', command='superuser-init-done') await verify_root_bridge_running(bridge, transport) @pytest.mark.asyncio async def test_no_login_messages(transport: MockTransport) -> None: await transport.check_bus_call('/LoginMessages', 'cockpit.LoginMessages', 'Get', [], ["{}"]) await transport.check_bus_call('/LoginMessages', 'cockpit.LoginMessages', 'Dismiss', [], []) await transport.check_bus_call('/LoginMessages', 'cockpit.LoginMessages', 'Get', [], ["{}"]) @pytest.fixture def login_messages_envvar(monkeypatch): if sys.version_info < (3, 8): pytest.skip("os.memfd_create new in 3.8") fd = os.memfd_create('login messages') os.write(fd, b"msg") # this is questionable (since it relies on ordering of fixtures), but it works monkeypatch.setenv('COCKPIT_LOGIN_MESSAGES_MEMFD', str(fd)) return None @pytest.mark.asyncio async def test_login_messages(login_messages_envvar, transport: MockTransport) -> None: del login_messages_envvar await transport.check_bus_call('/LoginMessages', 'cockpit.LoginMessages', 'Get', [], ["msg"]) # repeated read should get the messages again await transport.check_bus_call('/LoginMessages', 'cockpit.LoginMessages', 'Get', [], ["msg"]) # ...but not after they were dismissed await transport.check_bus_call('/LoginMessages', 'cockpit.LoginMessages', 'Dismiss', [], []) await transport.check_bus_call('/LoginMessages', 'cockpit.LoginMessages', 'Get', [], ["{}"]) # idempotency await transport.check_bus_call('/LoginMessages', 'cockpit.LoginMessages', 'Dismiss', [], []) await transport.check_bus_call('/LoginMessages', 'cockpit.LoginMessages', 'Get', [], ["{}"]) @pytest.mark.asyncio async def test_freeze(bridge: Bridge, transport: MockTransport) -> None: koelle = await transport.check_open('echo') malle = await transport.check_open('echo') # send a bunch of data to frozen koelle bridge.open_channels[koelle].freeze_endpoint() transport.send_data(koelle, b'x1') transport.send_data(koelle, b'x2') transport.send_data(koelle, b'x3') transport.send_done(koelle) # malle never freezes transport.send_data(malle, b'yy') transport.send_done(malle) # unfreeze koelle bridge.open_channels[koelle].thaw_endpoint() # malle should have sent its messages first await transport.assert_data(malle, b'yy') await transport.assert_msg('', command='done', channel=malle) await transport.assert_msg('', command='close', channel=malle) # the data from koelle should still be in the right order, though await transport.assert_data(koelle, b'x1') await transport.assert_data(koelle, b'x2') await transport.assert_data(koelle, b'x3') await transport.assert_msg('', command='done', channel=koelle) await transport.assert_msg('', command='close', channel=koelle) @pytest.mark.asyncio async def test_internal_metrics(transport: MockTransport) -> None: metrics = [ {"name": "cpu.core.user", "derive": "rate"}, {"name": "memory.used"}, ] interval = 100 source = 'internal' await transport.check_open('metrics1', source=source, interval=interval, metrics=metrics) _, data = await transport.next_frame() # first message is always the meta message meta = json.loads(data) assert isinstance(meta['timestamp'], float) assert meta['interval'] == interval assert meta['source'] == source assert isinstance(meta['metrics'], list) instances = len(next(m['instances'] for m in meta['metrics'] if m['name'] == 'cpu.core.user')) # actual data _, data = await transport.next_frame() data = json.loads(data) assert isinstance(data, list) # cpu.core.user instances should be the same as meta sent instances assert instances == len(data[0][0]) # all instances should be False, as this is a rate assert all(d is False for d in data[0][0]) # memory.used should be an integer assert isinstance(data[0][1], int) # next data for rate should not be False _, data = await transport.next_frame() data = json.loads(data) assert all(d is not False for d in data[0][0]) @pytest.mark.asyncio async def test_fsread1_errors(transport: MockTransport, tmp_path: Path) -> None: if os.geteuid() != 0: await transport.check_open('fsread1', path='/etc/shadow', problem='access-denied') else: print('running as root, skipping access-denied check') await transport.check_open('fsread1', path='/', problem='internal-error', reply_keys={'message': "[Errno 21] Is a directory: '/'"}) await transport.check_open('fsread1', path='/etc/passwd', max_read_size="lol", problem='protocol-error', reply_keys={'message': "attribute 'max_read_size': must have type int"}) await transport.check_open('fsread1', path='/etc/passwd', max_read_size=1, problem='too-large') # default read size limit big_file = tmp_path / 'bigfile.img' fd = os.open(big_file, os.O_RDWR | os.O_CREAT) os.posix_fallocate(fd, 0, 17 * 1024 * 1024) os.close(fd) await transport.check_open('fsread1', path=str(big_file), problem='too-large') # -1 implies no limit await transport.check_open('fsread1', path=str(big_file), max_read_size=-1) @pytest.mark.asyncio async def test_fsread1_size_hint(transport: MockTransport) -> None: data = None stat = os.stat('/usr/lib/os-release') with open('/usr/lib/os-release', 'rb') as fp: data = fp.read() ch = await transport.check_open('fsread1', path='/usr/lib/os-release', binary='raw', reply_keys={'size-hint': stat.st_size}) await transport.assert_data(ch, data) @pytest.mark.asyncio async def test_fsread1_size_hint_absent(transport: MockTransport) -> None: # non-binary fsread1 has no size-hint await transport.check_open('fsread1', path='/etc/passwd', absent_keys=['size-hint']) @pytest.mark.asyncio async def test_fsread1_size_hint_absent_char_device(transport: MockTransport) -> None: # character device fsread1 has no size-hint await transport.check_open('fsread1', path='/dev/null', binary='raw', absent_keys=['size-hint']) @pytest.mark.asyncio async def test_fslist1_no_watch(transport: MockTransport, tmp_path: Path) -> None: # empty ch = await transport.check_open('fslist1', path=str(tmp_path), watch=False) await transport.assert_msg('', command='done', channel=ch) await transport.check_close(channel=ch) # create a file and a directory in some_dir (tmp_path / 'somefile').touch() (tmp_path / 'somedir').mkdir() ch = await transport.check_open('fslist1', path=str(tmp_path), watch=False) # don't assume any ordering msg1 = await transport.next_msg(ch) msg2 = await transport.next_msg(ch) if msg1['type'] == 'file': msg1, msg2 = msg2, msg1 assert msg1 == {'event': 'present', 'path': 'somedir', 'type': 'directory'} assert msg2 == {'event': 'present', 'path': 'somefile', 'type': 'file'} await transport.assert_msg('', command='done', channel=ch) await transport.check_close(channel=ch) @pytest.mark.asyncio async def test_fslist1_notexist(transport: MockTransport) -> None: await transport.check_open( 'fslist1', path='/nonexisting', watch=False, problem='not-found', reply_keys={'message': "[Errno 2] No such file or directory: '/nonexisting'"}) class GrpPwMock: def __init__(self, uid: 'int', gid: 'int | None' = None) -> None: self.pw_uid = uid self.gr_gid = gid def __str__(self) -> str: return f'uid={self.pw_uid},gid={self.gr_gid}' @pytest.fixture def fchown_mock() -> Generator[unittest.mock.MagicMock, unittest.mock.MagicMock, None]: with unittest.mock.patch('os.fchown', return_value=True) as fchown_mock: yield fchown_mock @pytest.fixture def user_group_mock(user_group_mock_arg: GrpPwMock) -> Generator[GrpPwMock, GrpPwMock, None]: with unittest.mock.patch('pwd.getpwnam', return_value=user_group_mock_arg): with unittest.mock.patch('grp.getgrnam', return_value=user_group_mock_arg): yield user_group_mock_arg @pytest.mark.asyncio async def test_fsreplace1(transport: MockTransport, tmp_path: Path) -> None: # create non-existing file myfile = tmp_path / 'newfile' ch = await transport.check_open('fsreplace1', path=str(myfile)) transport.send_data(ch, b'some stuff') transport.send_done(ch) await transport.assert_msg('', command='done', channel=ch) await transport.check_close(channel=ch) assert myfile.read_bytes() == b'some stuff' # no leftover files assert os.listdir(tmp_path) == ['newfile'] # default umask is applied prev_umask = os.umask(0) os.umask(prev_umask) assert stat.S_IMODE(myfile.stat().st_mode) == 0o666 & ~prev_umask # now update its contents ch = await transport.check_open('fsreplace1', path=str(myfile)) transport.send_data(ch, b'new new new!') transport.send_done(ch) await transport.assert_msg('', command='done', channel=ch) await transport.check_close(channel=ch) assert myfile.read_bytes() == b'new new new!' # no leftover files assert os.listdir(tmp_path) == ['newfile'] # set funny permissions to check that they are preserved perms = 0o625 myfile.chmod(perms) # get the current tag ch = await transport.check_open('fsread1', path=str(myfile)) transport.send_done(ch) await transport.assert_data(ch, b'new new new!') await transport.assert_msg('', command='done', channel=ch) transport.send_close(ch) close_msg = await transport.next_msg('') assert close_msg['command'] == 'close' tag = close_msg['tag'] # update contents with expected tag ch = await transport.check_open('fsreplace1', path=str(myfile), tag=tag) transport.send_data(ch, b'even newer') transport.send_done(ch) await transport.assert_msg('', command='done', channel=ch) await transport.check_close(channel=ch) assert myfile.read_bytes() == b'even newer' # preserves existing permissions when giving expected tag assert stat.S_IMODE(myfile.stat().st_mode) == perms # mode supplied via attrs overrides existing permissions new_perms = 0o422 ch = await transport.check_open('fsreplace1', path=str(myfile), attrs={'mode': new_perms}) transport.send_data(ch, b'in the future') transport.send_done(ch) await transport.assert_msg('', command='done', channel=ch) await transport.check_close(channel=ch) assert myfile.read_bytes() == b'in the future' assert stat.S_IMODE(myfile.stat().st_mode) == new_perms # write empty file ch = await transport.check_open('fsreplace1', path=str(myfile)) transport.send_data(ch, b'') transport.send_done(ch) await transport.assert_msg('', command='done', channel=ch) await transport.check_close(channel=ch) assert myfile.read_bytes() == b'' # delete file ch = await transport.check_open('fsreplace1', path=str(myfile)) transport.send_done(ch) await transport.assert_msg('', command='done', channel=ch) await transport.check_close(channel=ch) assert not myfile.exists() # acks ch = await transport.check_open('fsreplace1', path=str(myfile), send_acks='bytes') transport.send_data(ch, b'some stuff') await transport.assert_msg('', command='ack', bytes=10, channel=ch) transport.send_data(ch, b'some more stuff') await transport.assert_msg('', command='ack', bytes=15, channel=ch) transport.send_done(ch) await transport.assert_msg('', command='done', channel=ch) await transport.check_close(channel=ch) # tag is "-" and file should not exist newfile = tmp_path / 'new' ch = await transport.check_open('fsreplace1', path=str(newfile), tag='-') transport.send_data(ch, b'some stuff') transport.send_done(ch) await transport.assert_msg('', command='done', channel=ch) await transport.check_close(channel=ch) assert newfile.read_text() == 'some stuff' # delete the now written newfile with tag='-' the file should now exist so this should fail ch = await transport.check_open('fsreplace1', path=str(newfile), tag='-') transport.send_done(ch) await transport.assert_msg('', problem='change-conflict', channel=ch) assert newfile.exists() # create file with a custom mode newfile.unlink() assert not newfile.exists() ch = await transport.check_open('fsreplace1', path=str(newfile), attrs={'mode': perms}) transport.send_data(ch, b'some stuff') transport.send_done(ch) await transport.assert_msg('', command='done', channel=ch) await transport.check_close(channel=ch) assert newfile.read_text() == 'some stuff' assert stat.S_IMODE(newfile.stat().st_mode) == perms # create file with tag='-' and custom mode newfile.unlink() assert not newfile.exists() ch = await transport.check_open('fsreplace1', path=str(newfile), tag='-', attrs={'mode': perms}) transport.send_data(ch, b'some stuff') transport.send_done(ch) await transport.assert_msg('', command='done', channel=ch) await transport.check_close(channel=ch) assert newfile.read_text() == 'some stuff' assert stat.S_IMODE(newfile.stat().st_mode) == perms @pytest.mark.asyncio async def test_fsreplace1_change_conflict(transport: MockTransport, tmp_path: Path) -> None: myfile = tmp_path / 'data' myfile.write_text('hello') # get current tag from fsread1 ch = await transport.check_open('fsread1', path=str(myfile)) transport.send_done(ch) await transport.assert_data(ch, b'hello') await transport.assert_msg('', command='done', channel=ch) transport.send_close(ch) close_msg = await transport.next_msg('') assert close_msg['command'] == 'close' tag = close_msg['tag'] # modify the file in between read and replace operations # we have to wait a bit, assuming that file systems we run tests on have at least centisecond mtime resolution await asyncio.sleep(0.2) myfile.write_text('goodbye') # try to replace it, expecting the old contents (via tag) ch = await transport.check_open('fsreplace1', path=str(myfile), tag=tag) transport.send_data(ch, b'newcontent') transport.send_done(ch) await transport.assert_msg('', command='close', channel=ch, problem='change-conflict') transport.send_close(ch) # file was not touched by fsreplace1 due to conflict assert myfile.read_text() == 'goodbye' @pytest.mark.asyncio async def test_fsreplace1_change_conflict_mode(transport: MockTransport, tmp_path: Path) -> None: myfile = tmp_path / 'data' myfile.write_text('hello') # get current tag ch = await transport.check_open('fsread1', path=str(myfile)) transport.send_done(ch) await transport.assert_data(ch, b'hello') await transport.assert_msg('', command='done', channel=ch) transport.send_close(ch) close_msg = await transport.next_msg('') assert close_msg['command'] == 'close' tag = close_msg['tag'] # modify the stat metadata in between read and replace operations new_mode = 0o741 assert stat.S_IMODE(myfile.stat().st_mode) != new_mode myfile.chmod(new_mode) # try to replace it, expecting the old mode (via tag) ch = await transport.check_open('fsreplace1', path=str(myfile), tag=tag) transport.send_data(ch, b'newcontent') transport.send_done(ch) await transport.assert_msg('', command='close', channel=ch, problem='change-conflict') transport.send_close(ch) # file was not touched by fsreplace1 due to conflict assert stat.S_IMODE(myfile.stat().st_mode) == new_mode assert myfile.read_text() == 'hello' @pytest.mark.asyncio async def test_fsreplace1_error(transport: MockTransport, tmp_path: Path) -> None: # trying to write a directory ch = await transport.check_open('fsreplace1', path=str(tmp_path)) transport.send_data(ch, b'not me') transport.send_done(ch) await transport.assert_msg('', command='close', channel=ch, problem='access-denied') # nonexisting directory ch = await transport.check_open('fsreplace1', path='/non/existing/file') transport.send_data(ch, b'not me') transport.send_done(ch) await transport.assert_msg('', command='close', channel=ch, problem='not-found') # invalid send-acks option await transport.check_open('fsreplace1', path=str(tmp_path), send_acks='not-valid', problem='protocol-error', reply_keys={ 'message': """attribute 'send-acks': invalid value "not-valid" not in ['bytes']""" }) await transport.check_open('fsreplace1', path=str(tmp_path / 'test'), attrs={'user': 'cockpit', 'group': 'cockpit', 'selinux': True}, problem='protocol-error', reply_keys={ 'message': '"attrs" contains unsupported key(s): selinux', 'unsupported_attrs': ['selinux'] }) await transport.check_open('fsreplace1', path=str(tmp_path / 'test'), attrs={'user': 'cockpit'}, problem='protocol-error', reply_keys={ 'message': '"group" attribute is empty while "user" is provided' }) await transport.check_open('fsreplace1', path=str(tmp_path / 'test'), attrs={'group': 'cockpit'}, problem='protocol-error', reply_keys={ 'message': '"user" attribute is empty while "group" is provided' }) await transport.check_open('fsreplace1', path=str(tmp_path / 'test'), attrs={'user': 'cockpit', 'group': []}, problem='protocol-error', reply_keys={ 'message': "attribute 'attrs': attribute 'group': must be a string or integer" }) await transport.check_open('fsreplace1', path=str(tmp_path / 'test'), attrs={'user': [], 'group': 'test'}, problem='protocol-error', reply_keys={ 'message': "attribute 'attrs': attribute 'user': must be a string or integer" }) mock_val = GrpPwMock(uid=0, gid=1000) with unittest.mock.patch('pwd.getpwnam', side_effect=KeyError()): await transport.check_open('fsreplace1', path=str(tmp_path / 'test'), attrs={'user': 'bazinga', 'group': 'foo'}, problem='not-found', reply_keys={ 'message': 'uid not found for bazinga' }) with unittest.mock.patch('pwd.getpwnam', return_value=mock_val): with unittest.mock.patch('grp.getgrnam', side_effect=KeyError()): await transport.check_open('fsreplace1', path=str(tmp_path / 'test'), attrs={'user': 'bazinga', 'group': 'test'}, problem='not-found', reply_keys={ 'message': 'gid not found for test' }) ATTRS_TEST_DATA = [ (GrpPwMock(1111, 1110), {'user': 1111, 'group': 1110}), (GrpPwMock(1111, 1110), {'user': 'monkey', 'group': 'group'}), ] @pytest.mark.parametrize(('user_group_mock_arg', 'attrs'), ATTRS_TEST_DATA) @pytest.mark.asyncio async def test_fsreplace1_attrs(transport: MockTransport, fchown_mock: unittest.mock.MagicMock, tmp_path: Path, user_group_mock, attrs: Dict[str, Union[int, str]]) -> None: async def existing_file_with_attrs(filename: str, tag: str) -> None: test_file = tmp_path / filename ch = await transport.check_open('fsreplace1', path=str(test_file), attrs=attrs, tag=tag) transport.send_data(ch, b'content') transport.send_done(ch) await transport.assert_msg('', command='done', channel=ch) await transport.check_close(ch) assert test_file.read_bytes() == b'content' async def create_file_with_attrs(filename: str) -> None: test_file = tmp_path / filename ch = await transport.check_open('fsreplace1', path=str(test_file), attrs=attrs) transport.send_data(ch, b'content') transport.send_done(ch) await transport.assert_msg('', command='done', channel=ch) await transport.check_close(ch) assert test_file.read_bytes() == b'content' await create_file_with_attrs('test') fchown_mock.assert_called() _, call_uid, call_gid = fchown_mock.call_args[0] assert call_uid == user_group_mock.pw_uid if user_group_mock.gr_gid is None: assert call_gid == user_group_mock.pw_uid else: assert call_gid == user_group_mock.gr_gid # Existing file existing_file = tmp_path / 'existing' existing_file.write_text('new') tag = tag_from_path(existing_file) assert tag is not None await existing_file_with_attrs('existing', tag) fchown_mock.assert_called() _, call_uid, call_gid = fchown_mock.call_args[0] assert call_uid == user_group_mock.pw_uid if user_group_mock.gr_gid is None: assert call_gid == user_group_mock.pw_uid else: assert call_gid == user_group_mock.gr_gid @pytest.mark.asyncio async def test_fsreplace1_size_hint(transport: MockTransport, tmp_path: Path) -> None: myfile = tmp_path / 'myfile' # create a file, no size hint ch = await transport.check_open('fsreplace1', path=str(myfile)) assert list(tmp_path.iterdir()) == [] # no temp file should be created yet transport.send_data(ch, b'content') transport.send_done(ch) await transport.assert_msg('', command='done', channel=ch) await transport.check_close(ch) assert list(tmp_path.iterdir()) == [myfile] # no temp file left assert myfile.read_bytes() == b'content' # delete it again ch = await transport.check_open('fsreplace1', path=str(myfile)) assert list(tmp_path.iterdir()) == [myfile] # no temp file created transport.send_done(ch) await transport.assert_msg('', command='done', channel=ch) await transport.check_close(ch) assert list(tmp_path.iterdir()) == [] # empty again # enospc in advance: 1PB ought to be enough for anybody... ch = transport.send_open('fsreplace1', path=str(myfile), size=1 << 50) # 1PB err = await transport.assert_msg('', command='close', channel=ch, problem='internal-error') message = get_str(err, 'message') assert 'No space left on device' in message or 'File too large' in message assert list(tmp_path.iterdir()) == [] # still empty assert not myfile.exists() # ensure cancellation is always a no-op, even with size specified ch = await transport.check_open('fsreplace1', path=str(myfile), size=1 << 20) # 1MB assert len(list(tmp_path.iterdir())) == 1 # observe the temp file created immediately await transport.check_close(ch) assert list(tmp_path.iterdir()) == [] # no temp file left assert not myfile.exists() # ...but even size=0 will result in the file being created without payload ch = await transport.check_open('fsreplace1', path=str(myfile), size=0) assert len(list(tmp_path.iterdir())) == 1 # observe the temp file created immediately transport.send_done(ch) await transport.assert_msg('', command='done', channel=ch) await transport.check_close(ch) assert list(tmp_path.iterdir()) == [myfile] # no temp file left assert myfile.read_bytes() == b'' # check too little data written ch = await transport.check_open('fsreplace1', path=str(myfile), size=1 << 20) # 1 MB assert len(list(tmp_path.iterdir())) == 2 # observe the temp file created immediately transport.send_data(ch, b'too small') transport.send_done(ch) await transport.assert_msg('', command='done', channel=ch) await transport.check_close(ch) assert list(tmp_path.iterdir()) == [myfile] # no temp file left assert myfile.read_bytes() == b'too small' # check too much data written ch = await transport.check_open('fsreplace1', path=str(myfile), size=5) assert len(list(tmp_path.iterdir())) == 2 # observe the temp file created immediately transport.send_data(ch, b'too large') transport.send_done(ch) await transport.assert_msg('', command='done', channel=ch) await transport.check_close(ch) assert list(tmp_path.iterdir()) == [myfile] # no temp file left assert myfile.read_bytes() == b'too large' @pytest.mark.asyncio @pytest.mark.parametrize('channeltype', CHANNEL_TYPES) async def test_channel(bridge: Bridge, transport: MockTransport, channeltype, tmp_path: Path) -> None: payload = channeltype.payload args = dict(channeltype.restrictions) async def serve_page(reader, writer): while True: line = await reader.readline() if line.strip(): print('HTTP Request line', line) else: break print('Sending HTTP reply') writer.write(b'HTTP/1.1 200 OK\r\n\r\nA document\r\n') await writer.drain() writer.close() srv = str(tmp_path / 'sock') await asyncio.start_unix_server(serve_page, srv) if payload == 'fsinfo': args = {'path': '/', 'attrs': []} elif payload == 'fslist1': args = {'path': '/', 'watch': False} elif payload == 'fsread1': args = {'path': '/etc/passwd'} elif payload == 'fsreplace1': args = {'path': 'tmpfile'} elif payload == 'fswatch1': args = {'path': '/etc'} elif payload == 'http-stream1': args = {'internal': 'packages', 'method': 'GET', 'path': '/manifests.js', 'headers': {'X-Forwarded-Proto': 'http', 'X-Forwarded-Host': 'localhost'}} elif payload == 'http-stream2': args = {'method': 'GET', 'path': '/bzzt', 'unix': srv} elif payload == 'stream': if 'spawn' in args: args = {'spawn': ['cat']} else: args = {'unix': srv} elif payload == 'metrics1' and channeltype.restrictions: args['metrics'] = [{'name': 'memory.free'}] elif payload == 'metrics1': pytest.skip('no PCP metric data') elif payload == 'dbus-json3': if not os.path.exists('/run/dbus/system_bus_socket'): pytest.skip('no dbus') elif payload == 'session-control': pytest.skip('too special') else: args = {} print('sending open', payload, args) ch = transport.send_open(payload, **args) saw_data = False while True: channel, msg = await transport.next_frame() print(channel, msg) if channel == '': control = json.loads(msg) assert control['channel'] == ch command = control['command'] if command == 'ready': if 'spawn' in args: assert isinstance(control['pid'], int) with open(f"/proc/{control['pid']}/cmdline") as f: assert f.read() == 'cat\0' # If we get ready, it's our turn to send data first. # Hopefully we didn't receive any before. assert not saw_data break else: pytest.fail(f'unexpected event: {(payload, args, control)}') else: saw_data = True # If we're here, it's our turn to talk. Say nothing. print('sending done') transport.send_done(ch) if payload in ['dbus-json3', 'fswatch1', 'metrics1', 'null']: transport.send_close(ch) while True: channel, msg = await transport.next_frame() print(channel, msg) if channel == '': control = json.loads(msg) command = control['command'] if command == 'done': continue elif command == 'close': assert 'problem' not in control return @pytest.mark.parametrize(('os_release', 'expected'), [ # simple values, with comments and ignored space ( '\n\n# simple\nID=mylinux\nVERSION=1.2\n\n# comment\nFOO=bar\n\n', {'ID': 'mylinux', 'VERSION': '1.2', 'FOO': 'bar'} ), # quoted values ( '''SINGLE='foo:bar '\nDOUBLE=" bar//foo"\n''', {'SINGLE': 'foo:bar ', 'DOUBLE': ' bar//foo'} ), # ignore ungrammatical lines ( 'A=a\nNOVALUE\nDOUBLEEQ=a=b\nB=b', {'A': 'a', 'B': 'b'} ), # invalid values; anything outside [A-Za-z0-9] must be quoted; but our parser is more liberal ( 'X=a:b\nY=a b\nZ=a-b\nV=a_b', {'X': 'a:b', 'Z': 'a-b', 'V': 'a_b'} ), ]) def test_get_os_release(os_release: str, expected: str) -> None: with unittest.mock.patch('builtins.open', unittest.mock.mock_open(read_data=os_release)): assert Bridge.get_os_release() == expected class AckChannel(AsyncChannel): payload = 'ack1' async def run(self, options: JsonObject) -> None: self.semaphore = asyncio.Semaphore(0) self.ready() while await self.read(): await self.semaphore.acquire() @pytest.mark.asyncio async def test_async_acks(bridge: Bridge, transport: MockTransport) -> None: # Inject our mock channel type for rule in bridge.routing_rules: if isinstance(rule, ChannelRoutingRule): rule.table['ack1'] = [AckChannel] # invalid send-acks values await transport.check_open('ack1', send_acks=True, problem='protocol-error') await transport.check_open('ack1', send_acks='x', problem='protocol-error') # open the channel with acks off ch = await transport.check_open('ack1') # send a bunch of data and get no acks for _ in range(20): transport.send_data(ch, b'x') # this will assert that we receive only the close message (and no acks) await transport.check_close(ch) # open the channel with acks on ch = await transport.check_open('ack1', send_acks='bytes') # send a bunch of data for _ in range(20): transport.send_data(ch, b'x') # we should get exactly one ack (from the first read) before things block await transport.assert_msg('', channel=ch, command='ack', bytes=1) # this will assert that we receive only the close message (and no additional acks) await transport.check_close(ch) # open the channel with acks on ch = await transport.check_open('ack1', send_acks='bytes') # fish the open channel out of the bridge ack = bridge.open_channels[ch] assert isinstance(ack, AckChannel) # let's give ourselves a bit more headroom for _ in range(5): ack.semaphore.release() # send a bunch of data and get some acks for _ in range(10): transport.send_data(ch, b'x') for _ in range(6): await transport.assert_msg('', channel=ch, command='ack', bytes=1) # make sure that as we "consume" the data we get more acks: for _ in range(4): # no ack in the queue... await transport.assert_empty() ack.semaphore.release() # ... but now there is. await transport.assert_msg('', channel=ch, command='ack', bytes=1) # make some more room (for data we didn't send) for _ in range(5): ack.semaphore.release() # but we shouldn't have gotten any acks for those await transport.check_close(ch) @pytest.mark.asyncio async def test_flow_control(transport: MockTransport, tmp_path: Path) -> None: bigun = tmp_path / 'bigun' total_bytes = 8 * 1024 * 1024 recvd_bytes = 0 bigun.write_bytes(b'0' * total_bytes) fsread1 = await transport.check_open('fsread1', path=str(bigun), flow_control=True) # We should receive a number of blocks of initial data, each with a ping. # We save the pings to reply later. pings: 'deque[JsonObject]' = deque() async def recv_one(): nonlocal recvd_bytes channel, data = await transport.next_frame() assert channel == fsread1 assert data == b'0' * Channel.BLOCK_SIZE recvd_bytes += len(data) ping = await transport.next_msg('') assert ping['command'] == 'ping' assert ping['channel'] == fsread1 assert ping['sequence'] == recvd_bytes pings.append(ping) while recvd_bytes < Channel.SEND_WINDOW: await recv_one() # We should stall out here. Make sure nothing else arrives. await transport.assert_empty() # Start sending pongs and make sure we receive a new block of data for each # one (as the window extends) while recvd_bytes < total_bytes: ping = pings.popleft() transport.send_json('', **dict(ping, command='pong')) await recv_one() transport.send_close(fsread1) @pytest.mark.asyncio async def test_large_upload(transport: MockTransport, tmp_path: Path) -> None: fifo = str(tmp_path / 'pipe') os.mkfifo(fifo) sender = await transport.check_open('stream', spawn=['dd', f'of={fifo}']) # cockpit.js doesn't do flow control, so neither do we... chunk = b'0' * Channel.BLOCK_SIZE loops = 100 for _ in range(loops): transport.send_data(sender, chunk) transport.send_done(sender) # we should be in a state now where we have a bunch of bytes queued up in # the bridge but they can't be delivered because nobody is reading from the # pipe... make sure dd is still running and we didn't get any messages. await transport.assert_empty() # start draining now, and make sure we get everything we sent. with open(fifo, 'rb') as receiver: received = await asyncio.get_running_loop().run_in_executor(None, receiver.read) assert len(received) == loops * Channel.BLOCK_SIZE # and now our done and close messages should come await transport.assert_msg('', command='done', channel=sender) await transport.assert_msg('', command='close', channel=sender) class FsInfoClient: def __init__(self, transport: MockTransport, channel: str): self.transport = transport self.channel = channel self.state: JsonObject = {} @classmethod async def open( cls, transport: MockTransport, path: 'str | Path', attrs: Sequence[str] = ('type', 'target'), fnmatch: str = '*.txt', **kwargs: JsonValue ) -> 'FsInfoClient': channel = await transport.check_open('fsinfo', path=str(path), attrs=attrs, fnmatch=fnmatch, reply_keys=None, absent_keys=(), **kwargs) # type: ignore[arg-type] return cls(transport, channel) async def next_state(self) -> JsonObject: while True: patch = await self.transport.next_msg(self.channel) self.state = json_merge_patch(self.state, patch) if not get_bool(self.state, 'partial', None): break return self.state async def wait(self) -> JsonObject: state = await self.next_state() await self.transport.assert_msg('', command='done', channel=self.channel) await self.transport.assert_msg('', command='close', channel=self.channel) return state async def close(self): await self.transport.check_close(self.channel) def fsinfo_err(err: int) -> JsonObject: problems = { errno.ENOENT: 'not-found', errno.EPERM: 'access-denied', errno.EACCES: 'access-denied', errno.ENOTDIR: 'not-directory' } return { 'error': { 'problem': problems.get(err, 'internal-error'), 'message': os.strerror(err), 'errno': errno.errorcode[err], } } @pytest.fixture def fsinfo_test_cases(tmp_path: Path) -> 'Iterator[dict[Path, JsonObject]]': # a normal directory normal_dir = tmp_path / 'dir' normal_dir.mkdir() (normal_dir / 'dir-file.txt').write_text('dir file') (normal_dir / 'dir-file.xtx').write_text('do not read this') # a directory without +x (search) no_x_dir = tmp_path / 'no-x-dir' no_x_dir.mkdir() (no_x_dir / 'no-x-dir-file.txt').write_text('file') no_x_dir.chmod(0o644) # a directory without +r (read) no_r_dir = tmp_path / 'no-r-dir' no_r_dir.mkdir() (no_r_dir / 'no-r-dir-file.txt').write_text('file') no_r_dir.chmod(0o311) # a normal file file = tmp_path / 'file' file.write_text('normal file') # a non-readable file no_r_file = tmp_path / 'no-r-file' no_r_file.write_text('inaccessible file') no_r_file.chmod(0) # a device dev = tmp_path / 'dev' dev.symlink_to('/dev/null') # a dangling symlink dangling = tmp_path / 'dangling' dangling.symlink_to('does-not-exist') # a symlink pointing to itself loopy = tmp_path / 'loopy' loopy.symlink_to(loopy) expected_state: 'dict[Path, JsonObject]' = { normal_dir: {"info": {"type": "dir", "entries": {'dir-file.txt': {"type": "reg"}}}}, # can't stat() the file no_x_dir: {"info": {"type": "dir", "entries": {"no-x-dir-file.txt": {}}}}, # can't read the directory, so no entries no_r_dir: {"info": {"type": "dir"}}, # normal file, can read its contents file: {"info": {"type": "reg"}}, # can't read file, so no contents no_r_file: {"info": {"type": "reg"}}, # a device dev: {"info": {"type": "chr"}}, # a dangling symlink dangling: fsinfo_err(errno.ENOENT), # a link pointing at itself loopy: fsinfo_err(errno.ELOOP), } if os.getuid() == 0: # we can't do the permissions-dependent tests as root del expected_state[no_x_dir] del expected_state[no_r_dir] del expected_state[no_r_file] try: yield expected_state finally: # restore writable permissions on directories. pytest has trouble cleaning # these up, otherwise... for path in expected_state: if not path.is_symlink() and path.is_dir(): path.chmod(0o700) @pytest.mark.asyncio async def test_fsinfo_nopath(transport: MockTransport) -> None: await transport.check_open('fsinfo', attrs=['type'], problem='protocol-error') @pytest.mark.asyncio async def test_fsinfo_noattrs(transport: MockTransport) -> None: await transport.check_open('fsinfo', path='/', problem='protocol-error') @pytest.mark.asyncio async def test_fsinfo_relative(transport: MockTransport) -> None: await transport.check_open('fsinfo', path='rel', problem='protocol-error') await transport.check_open('fsinfo', path='.', problem='protocol-error') @pytest.mark.asyncio async def test_fsinfo_nofollow_watch(transport: MockTransport) -> None: await transport.check_open('fsinfo', path='/', attrs=[], watch=True, follow=False, problem='protocol-error') @pytest.mark.asyncio async def test_fsinfo_nofollow_targets(transport: MockTransport) -> None: await transport.check_open('fsinfo', path='/', attrs=['targets'], follow=False, problem='protocol-error') @pytest.mark.asyncio async def test_fsinfo_empty_update(transport: MockTransport, tmp_path: Path) -> None: # test an empty update — make sure nothing lands on the wire ch = await transport.check_open('fsinfo', path=str(tmp_path), attrs=['type'], watch=True) assert await transport.next_msg(ch) == {'info': {"type": "dir"}} tmp_path.touch() await asyncio.sleep(0.1) # fsinfo waits 0.1 before dispatching updates await transport.assert_empty() # this waits another 0.1 await transport.check_close(ch) @pytest.mark.asyncio async def test_fsinfo(transport: MockTransport, fsinfo_test_cases: 'dict[Path, JsonObject]') -> None: for path, expected_state in fsinfo_test_cases.items(): client = await FsInfoClient.open(transport, path) assert await client.wait() == expected_state @pytest.mark.asyncio async def test_fsinfo_nofollow(transport: MockTransport, fsinfo_test_cases: 'dict[Path, JsonObject]') -> None: for path, expected_state in fsinfo_test_cases.items(): if path.name == 'loopy': # with nofollow, this won't fail — we'll see the link itself expected_state = {"info": {"type": "lnk", "target": str(path)}} elif path.name == 'dev': expected_state = {"info": {"type": "lnk", "target": "/dev/null"}} elif path.name == 'dangling': expected_state = {"info": {"type": "lnk", "target": "does-not-exist"}} client = await FsInfoClient.open(transport, path, follow=False) assert await client.wait() == expected_state @pytest.mark.asyncio async def test_fsinfo_onlydir(transport: MockTransport, fsinfo_test_cases: 'dict[Path, JsonObject]') -> None: for path, expected_state in fsinfo_test_cases.items(): if 'dir' not in path.name and 'error' not in expected_state: expected_state = fsinfo_err(errno.ENOTDIR) # with '/' appended, this should only open dirs client = await FsInfoClient.open(transport, str(path) + '/') assert await client.wait() == expected_state, f'for {path}' @pytest.mark.asyncio async def test_fsinfo_onlydir_watch(transport: MockTransport, fsinfo_test_cases: 'dict[Path, JsonObject]') -> None: for path, expected_state in fsinfo_test_cases.items(): # note the order here: because our check for notdir is implemented # inside of the bridge, we try inotify first and check for notdir # second. if systemd_ctypes supports this some day, it may change. if path.name.startswith('no-r'): # we need this one because we can't inotify files without +r expected_state = fsinfo_err(errno.EACCES) elif 'dir' not in path.name and 'error' not in expected_state: # and this one to deal with not-dir expected_state = fsinfo_err(errno.ENOTDIR) # with '/' appended, this should only open dirs client = await FsInfoClient.open(transport, str(path) + '/', watch=True) assert await client.next_state() == expected_state await client.close() @pytest.mark.asyncio async def test_fsinfo_watch_identity_changes( transport: MockTransport, tmp_path: Path, fsinfo_test_cases: 'dict[Path, JsonObject]' ) -> None: # we will now point a symlink to the various possibilities to make sure the # transitions are correctly handled link = tmp_path / 'link' client = await FsInfoClient.open(transport, link, watch=True) # we didn't make a link yet, so... assert (await client.next_state()) == fsinfo_err(errno.ENOENT) # in watch mode we report errors a bit more sensitively: we also report # them if we can't inotify the inode in question, which requires +r # permission. as such, we need to modify our 'no-r' tests: for path in list(fsinfo_test_cases): if path.name.startswith('no-r'): fsinfo_test_cases[path] = fsinfo_err(errno.EACCES) if path.name == 'dangling': # this breaks our logic below because we transition from ENOENT to ENOENT del fsinfo_test_cases[path] possibilities = tuple(fsinfo_test_cases) for from_file in possibilities: for to_file in possibilities: if from_file is to_file: continue # link to the original file and check state link.symlink_to(from_file) assert await client.next_state() == fsinfo_test_cases[from_file] # Try to generate some events immediately before we switch the link # to ensure the event is suppressed. Any one of these could fail # due to not being a directory or not having permissions. with contextlib.suppress(OSError): from_file.touch() with contextlib.suppress(OSError): (from_file / 'a.txt').touch() with contextlib.suppress(OSError): (from_file / 'a.txt').unlink() # atomic replace, check state (tmp_path / 'tmp').symlink_to(to_file) (tmp_path / 'tmp').rename(link) assert await client.next_state() == fsinfo_test_cases[to_file] if to_file.name == 'dir': # copied from fsinfo_test_cases fixture dir_entries = {'dir-file.txt': {"type": "reg"}} expected_state = {"info": {"type": "dir", "entries": dir_entries}} (to_file / 'a.txt').write_text('a file') dir_entries['a.txt'] = {"type": "reg"} (to_file / 'b.txt').write_text('b file') dir_entries['b.txt'] = {"type": "reg"} (to_file / 'a.xtx').write_text('b file') (to_file / 'b.xtx').write_text('b file') assert await client.next_state() == expected_state (to_file / 'a.xtx').unlink() (to_file / 'b.xtx').unlink() (to_file / 'dir.txt').mkdir() dir_entries['dir.txt'] = {"type": "dir"} (to_file / 'sym.txt').symlink_to('/x') dir_entries['sym.txt'] = {"type": "lnk", "target": "/x"} assert await client.next_state() == expected_state (to_file / 'sym.txt').unlink() del dir_entries['sym.txt'] assert await client.next_state() == expected_state # we intentionally try not to receive these events — we want # them to get lost when the identity changes, below (to_file / 'dir.txt').rmdir() (to_file / 'a.txt').unlink() (to_file / 'b.txt').unlink() # delete link to start over link.unlink() assert (await client.next_state()) == fsinfo_err(errno.ENOENT) # let the pending event handler occasionally run to find nothing to process await asyncio.sleep(0.15) await client.close() @pytest.mark.asyncio async def test_fsinfo_self_owner(transport: MockTransport, tmp_path: Path) -> None: client = await FsInfoClient.open(transport, tmp_path, ['user', 'uid', 'group', 'gid']) state = await client.wait() info = get_dict(state, 'info') assert get_int(info, 'uid') == os.getuid() assert get_int(info, 'gid') == os.getgid() assert info.get('user') == getpass.getuser() assert info.get('group') == grp.getgrgid(os.getgid()).gr_name # hopefully true... @pytest.mark.asyncio async def test_fsinfo_other_owner(transport: MockTransport, tmp_path: Path) -> None: tmpfile = tmp_path / 'x' tmpfile.touch() # try to get root to own this thing using a couple of tricks that may work # inside or outside of toolbox or containers quoted = shlex.quote(str(tmpfile)) subprocess.run(fr''' podman unshare chown 888:888 {quoted} || SUDO_ASKPASS=true sudo -A chown 0:0 '{quoted}' ''', shell=True, check=False) # verify that we ended up with a uid/gid with no user buf = tmpfile.stat() try: pwd.getpwuid(buf.st_uid) pytest.skip('Failed to find unmapped uid') except KeyError: pass # good! try: grp.getgrgid(buf.st_gid) pytest.skip('Failed to find unmapped gid') except KeyError: pass # good! client = await FsInfoClient.open(transport, tmpfile, ['user', 'uid', 'group', 'gid']) state = await client.wait() info = get_dict(state, 'info') assert get_int(info, 'uid') == buf.st_uid assert get_int(info, 'gid') == buf.st_gid assert info.get('user') == buf.st_uid # numeric fallback assert info.get('group') == buf.st_gid # numeric fallback @pytest.mark.asyncio async def test_fsinfo_targets(transport: MockTransport, tmp_path: Path) -> None: # we are only interested in the things that start with 'l' watch = await FsInfoClient.open(transport, tmp_path, ['type', 'target', 'targets'], fnmatch='l*', watch=True) entries: JsonDict = {} targets: JsonDict = {} state = {"info": {"type": "dir", "entries": entries, "targets": targets}} assert await watch.next_state() == state # none of those will show up in entries (not 'l*') (tmp_path / 'dir').mkdir() (tmp_path / 'dir' / 'file').write_text('abc') (tmp_path / 'dir' / 'lonely').write_text('abc') (tmp_path / 'dir' / 'dir').mkdir() (tmp_path / 'file').write_text('abc') (tmp_path / 'Lonely').write_text('abc') # this one will show up in entries because it matches 'l*' (tmp_path / 'loved').write_text('abc') entries['loved'] = {'type': 'reg'} # a link that won't show up anywhere (no fnmatch) (tmp_path / 'LonelyLink').symlink_to('dir/lonely') # link to things that will land in targets because they're not in fnmatch (tmp_path / 'lfile').symlink_to('file') entries['lfile'] = {'type': 'lnk', 'target': 'file'} targets['file'] = {'type': 'reg'} (tmp_path / 'ldir').symlink_to('dir') entries['ldir'] = {'type': 'lnk', 'target': 'dir'} targets['dir'] = {'type': 'dir'} # link to things that will land in targets because they're in another dir (tmp_path / 'ldirfile').symlink_to('dir/file') entries['ldirfile'] = {'type': 'lnk', 'target': 'dir/file'} targets['dir/file'] = {'type': 'reg'} (tmp_path / 'ldirdir').symlink_to('dir/dir') entries['ldirdir'] = {'type': 'lnk', 'target': 'dir/dir'} targets['dir/dir'] = {'type': 'dir'} (tmp_path / 'lnull').symlink_to('/dev/null') entries['lnull'] = {'type': 'lnk', 'target': '/dev/null'} targets['/dev/null'] = {'type': 'chr'} (tmp_path / 'lroot').symlink_to('/') entries['lroot'] = {'type': 'lnk', 'target': '/'} targets['/'] = {'type': 'dir'} # link to things that won't land in targets because they're in entries (tmp_path / 'llfile').symlink_to('lfile') entries['llfile'] = {'type': 'lnk', 'target': 'lfile'} (tmp_path / 'lldir').symlink_to('ldir') entries['lldir'] = {'type': 'lnk', 'target': 'ldir'} (tmp_path / 'lloved').symlink_to('loved') entries['lloved'] = {'type': 'lnk', 'target': 'loved'} # a link to the current directory won't show up in targets since we already # report its attributes directly as our toplevel state structure (tmp_path / 'ldot').symlink_to('.') entries['ldot'] = {'type': 'lnk', 'target': '.'} # a link to the parent directory should definitely show up, though (tmp_path / 'ldotdot').symlink_to('..') entries['ldotdot'] = {'type': 'lnk', 'target': '..'} targets['..'] = {'type': 'dir'} # make sure the watch managed to pick that all up assert await watch.next_state() == state # double-check with the non-watch variant client = await FsInfoClient.open(transport, tmp_path, ['type', 'target', 'targets'], fnmatch='l*') assert await client.wait() == state @pytest.mark.asyncio async def test_fsinfo_access_attrs(transport: MockTransport, fsinfo_test_cases: 'dict[Path, JsonObject]') -> None: for path, expected_state in fsinfo_test_cases.items(): # these are errors if path.name == 'dangling' or path.name == 'loopy': continue read_ok = True write_ok = True exec_ok = path.is_dir() if path.name == 'no-r-dir': read_ok = False elif path.name == 'no-x-dir': exec_ok = False elif path.name == 'no-r-file': read_ok = False write_ok = False exec_ok = False expected_state = {'info': {'r-ok': read_ok, 'w-ok': write_ok, 'x-ok': exec_ok}} # fnmatch='' to not include entries client = await FsInfoClient.open(transport, path, attrs=['w-ok', 'r-ok', 'x-ok'], fnmatch='') assert await client.wait() == expected_state, f'for path={path.name}' # Symlink access bits are always allowed for path, expected_state in fsinfo_test_cases.items(): if path.name != 'dev': continue expected_state = {'info': {'r-ok': True, 'w-ok': True, 'x-ok': True}} client = await FsInfoClient.open(transport, path, attrs=['w-ok', 'r-ok', 'x-ok'], follow=False) assert await client.wait() == expected_state, f'for path={path.name}' @pytest.mark.parametrize('supports_askpass', [False, True], ids=['no-askpass', 'askpass']) def test_is_valid_superuser_config_sudo_askpass(tmp_path: Path, *, supports_askpass: bool) -> None: sudo = tmp_path / 'sudo' sudo.write_text(f'#!/bin/sh\nexit {"0" if supports_askpass else "1"}\n') sudo.chmod(0o755) config = BridgeConfig({ 'privileged': True, 'spawn': [str(sudo), '-k', '-A', 'cockpit-bridge', '--privileged'], }) assert is_valid_superuser_config(config) == supports_askpass def test_is_valid_superuser_config_not_sudo(tmp_path: Path) -> None: danger = tmp_path / 'danger' pkexec = tmp_path / 'pkexec' pkexec.write_text(f'#!/bin/sh\ntouch {danger}\n') pkexec.chmod(0o755) config = BridgeConfig({ 'privileged': True, 'spawn': [str(pkexec), 'cockpit-bridge', '--privileged'], }) assert is_valid_superuser_config(config) assert not danger.exists() @pytest.mark.asyncio @pytest.mark.parametrize('absolute', [False, True], ids=['relpath', 'abspath']) @pytest.mark.parametrize('custom_path', [False, True], ids=['default-path', 'custom-path']) @pytest.mark.parametrize('exists', [False, True], ids=['missing', 'exists']) async def test_is_valid_superuser_config( tmp_path: Path, *, absolute: bool, custom_path: bool, exists: bool ) -> None: args: JsonObject if absolute: args = {'spawn': ['/usr/bin/true' if exists else '/nonexistent/path/true']} elif custom_path: if exists: (tmp_path / 'my-bridge').write_text('#!/bin/sh\n') (tmp_path / 'my-bridge').chmod(0o755) args = {'spawn': ['my-bridge'], 'environ': [f'PATH={tmp_path}']} else: args = {'spawn': ['true' if exists else 'nonexistent-binary']} config = BridgeConfig({**args, 'privileged': True}) assert is_valid_superuser_config(config) == exists with contextlib.ExitStack() as expected: if not exists: expected.enter_context(pytest.raises(FileNotFoundError)) peer = ConfiguredPeer(unittest.mock.Mock(), config) await peer.do_connect_transport() peer.close()