77 lines
3.4 KiB
Python
77 lines
3.4 KiB
Python
import importlib.util
|
|
import json
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
spec = importlib.util.spec_from_file_location('verifier', Path(__file__).with_name('verify_agent_denials.py'))
|
|
v = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(v)
|
|
|
|
|
|
INVENTORY = {'cluster': 'railiance01', 'namespaces': ['default', 'sso'], 'complete': True, 'supervisor_receipt_ref': 'synthetic-test-only'}
|
|
|
|
|
|
class VerificationTest(unittest.TestCase):
|
|
def setUp(self):
|
|
# Retries back off between attempts; tests need no real delay.
|
|
sleeper = patch.object(v.time, 'sleep')
|
|
sleeper.start()
|
|
self.addCleanup(sleeper.stop)
|
|
|
|
def test_old_account_refuses_before_any_command(self):
|
|
with patch.object(v.os, 'geteuid', return_value=0), patch.object(v, 'run') as run:
|
|
self.assertFalse(v.probe('/kubectl', 'agent', 'custodian-agt', INVENTORY)['passed'])
|
|
run.assert_not_called()
|
|
|
|
def simulate(self, identity, denied_code=1, denied_output=b'no', allow_sso=False):
|
|
payload = json.dumps({'status': {'userInfo': identity}}).encode()
|
|
def response(argv):
|
|
if 'whoami' in argv:
|
|
return 0, payload
|
|
if allow_sso and '--namespace' in argv and argv[-1] == 'sso' and 'secrets' in argv:
|
|
return 0, b'yes'
|
|
return denied_code, denied_output
|
|
with patch.object(v.os, 'geteuid', return_value=1234), \
|
|
patch.object(v.pwd, 'getpwuid', return_value=SimpleNamespace(pw_name='custodian-agt')), \
|
|
patch.object(v.os, 'getgroups', return_value=[1234]), \
|
|
patch.object(v.os, 'getegid', return_value=1234), \
|
|
patch.object(v.grp, 'getgrgid', return_value=SimpleNamespace(gr_name='custodian-agt')), \
|
|
patch.object(v.os, 'access', return_value=False), \
|
|
patch.object(v, 'run', side_effect=response):
|
|
return v.probe('/kubectl', 'agent', 'custodian-agt', INVENTORY)
|
|
|
|
def test_admin_membership_refuses(self):
|
|
result = self.simulate({'username': v.IDENTITY, 'groups': ['system:masters']})
|
|
self.assertEqual(result['status'], 'refused_wrong_identity')
|
|
|
|
def test_timeout_is_not_denial(self):
|
|
result = self.simulate({'username': v.IDENTITY}, None, b'')
|
|
self.assertFalse(result['passed'])
|
|
|
|
def test_forbidden_text_is_not_valid_denial(self):
|
|
result = self.simulate({'username': v.IDENTITY}, 1, b'Forbidden secret material')
|
|
self.assertFalse(result['passed'])
|
|
self.assertNotIn('secret material', json.dumps(result))
|
|
|
|
def test_namespaced_allow_fails_even_when_all_namespace_query_would_deny(self):
|
|
result = self.simulate({'username': v.IDENTITY}, allow_sso=True)
|
|
self.assertFalse(result['passed'])
|
|
self.assertFalse(result['checks']['sso:get:secrets'])
|
|
|
|
def test_empty_inventory_refuses_before_commands(self):
|
|
with patch.object(v, 'run') as run:
|
|
result = v.probe('/kubectl', 'agent', 'custodian-agt', {**INVENTORY, 'namespaces': []})
|
|
self.assertFalse(result['passed'])
|
|
run.assert_not_called()
|
|
|
|
def test_negative_success_cannot_admit_assignment(self):
|
|
result = self.simulate({'username': v.IDENTITY})
|
|
self.assertTrue(result['passed'])
|
|
self.assertFalse(result['assignment_admitted'])
|
|
self.assertTrue(result['remaining'])
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|