53 lines
2.7 KiB
Python
53 lines
2.7 KiB
Python
"""Peer SSH enrollment preserves administrator keys and restricts the source."""
|
|
import base64
|
|
import importlib.util
|
|
import os
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
spec = importlib.util.spec_from_file_location('peer_key',
|
|
Path(__file__).resolve().parents[2] / 'scripts/install_ananke_peer_key.py')
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
|
|
|
|
class PeerKeyTests(unittest.TestCase):
|
|
def test_append_is_restricted_and_idempotent(self):
|
|
blob = b'\x00\x00\x00\x0bssh-ed25519\x00\x00\x00\x20' + b'B' * 32
|
|
payload = {'hostname': 'titan-test', 'public_key': 'ssh-ed25519 ' + base64.b64encode(blob).decode()}
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
home = Path(directory)
|
|
(home / '.ssh').mkdir()
|
|
keys = home / '.ssh/authorized_keys'
|
|
keys.write_text('# existing administrator key remains\nssh-ed25519 existing-admin\n')
|
|
account = SimpleNamespace(pw_dir=directory, pw_uid=os.getuid(), pw_gid=os.getgid())
|
|
with patch.object(module.os, 'geteuid', return_value=0), \
|
|
patch.object(module.socket, 'gethostname', return_value='titan-test'), \
|
|
patch.object(module.pwd, 'getpwnam', return_value=account):
|
|
self.assertTrue(module.install(payload)['key_added'])
|
|
self.assertTrue(module.install(payload)['key_already_present'])
|
|
content = keys.read_text()
|
|
self.assertIn('existing-admin', content)
|
|
self.assertEqual(content.count(payload['public_key']), 1)
|
|
self.assertIn('from="192.168.22.26",restrict ', content)
|
|
self.assertEqual(keys.stat().st_mode & 0o777, 0o600)
|
|
|
|
def test_wrong_node_is_rejected_before_account_lookup(self):
|
|
with patch.object(module.os, 'geteuid', return_value=0), \
|
|
patch.object(module.socket, 'gethostname', return_value='different-node'), \
|
|
patch.object(module.pwd, 'getpwnam') as lookup:
|
|
with self.assertRaisesRegex(ValueError, 'root_and_matching_hostname_required'):
|
|
module.install({'hostname': 'titan-test'})
|
|
lookup.assert_not_called()
|
|
|
|
def test_invalid_key_does_not_touch_authorized_keys(self):
|
|
with patch.object(module.os, 'geteuid', return_value=0), \
|
|
patch.object(module.socket, 'gethostname', return_value='titan-test'), \
|
|
patch.object(module.pwd, 'getpwnam') as lookup:
|
|
with self.assertRaises(ValueError):
|
|
module.install({'hostname': 'titan-test', 'public_key': 'ssh-ed25519 invalid'})
|
|
lookup.assert_not_called()
|