monitor_app/tests/test_authority.py

0001 """Authority: the truth table, the two gates, and the declarations.
0002
0003 docs/AUTHORITY.md. Plain assertions over the real modules; the database is
0004 the test database Django gives every TestCase.
0005 """
0006
0007 from django.contrib.auth import get_user_model
0008 from django.test import RequestFactory, TestCase, override_settings
0009
0010 from monitor_app import authority as A
0011 from monitor_app.middleware import (AUTHORITY_EXEMPT_URL_NAMES,
0012                                     AuthorityGateMiddleware,
0013                                     TunnelAuthentication)
0014 from monitor_app.models import SysConfig, UserPreference
0015
0016
0017 class AuthorityRuleTests(TestCase):
0018
0019     def test_may_act_truth_table(self):
0020         cases = [
0021             ({'eic': True, 'rights': None}, True),
0022             ({'eic': True, 'rights': 'read'}, False),
0023             ({'eic': False, 'rights': None}, False),
0024             ({'eic': False, 'rights': 'basic'}, True),
0025             ({'eic': None, 'rights': 'basic'}, True),
0026             ({'eic': None, 'rights': 'ops'}, True),
0027             ({'eic': None, 'rights': 'read'}, False),
0028             ({'eic': None, 'rights': None}, False),
0029         ]
0030         for record, expected in cases:
0031             self.assertEqual(A.may_act(record), expected, record)
0032         self.assertFalse(A.may_act('nobody-here'))
0033
0034     def test_setters_write_their_own_field_only(self):
0035         A.set_rights('u', 'basic')
0036         A.set_eic('u', False, github='u-gh')
0037         record = A.get_authority('u')
0038         self.assertEqual((record['eic'], record['rights'], record['github']),
0039                          (False, 'basic', 'u-gh'))
0040         A.set_eic('u', None)
0041         self.assertEqual(A.get_authority('u')['rights'], 'basic')
0042         with self.assertRaises(A.AuthorityError):
0043             A.set_rights('u', 'admin')
0044         with self.assertRaises(ValueError):
0045             UserPreference.set_pref('u', A.AUTHORITY_KEY, {})
0046
0047     def test_may_set_priority_needs_pac_or_ops_and_authority(self):
0048         cases = [
0049             ({'eic': True, 'rights': None, 'pac': False}, False),
0050             ({'eic': True, 'rights': None, 'pac': True}, True),
0051             ({'eic': False, 'rights': None, 'pac': True}, False),
0052             ({'eic': None, 'rights': 'ops', 'pac': False, 'ops': True}, True),
0053             ({'eic': True, 'rights': None, 'pac': False, 'ops': True}, True),
0054             ({'eic': None, 'rights': 'basic', 'pac': False}, False),
0055             ({'eic': None, 'rights': 'basic', 'pac': True}, True),
0056             ({'eic': True, 'rights': 'read', 'pac': True}, False),
0057         ]
0058         for record, expected in cases:
0059             self.assertEqual(A.may_set_priority(record), expected, record)
0060
0061     def test_roles_are_flags_beside_rights(self):
0062         A.set_rights('p', 'basic')
0063         A.set_pac('p', True)
0064         A.set_ops('p', True)
0065         record = A.get_authority('p')
0066         self.assertEqual((record['rights'], record['pac'], record['ops']),
0067                          ('basic', True, True))
0068         self.assertTrue(A.is_ops(record) and A.may_set_priority(record))
0069         A.set_pac('p', False)
0070         self.assertEqual(A.get_authority('p')['rights'], 'basic')
0071         self.assertFalse(A.get_authority('p')['pac'])
0072         self.assertTrue(A.get_authority('p')['ops'])
0073
0074     def test_failed_check_keeps_eic_and_clears_on_observation(self):
0075         A.set_eic('f', True, github='f-gh')
0076         A.set_check_failed('f', 'no stored GitHub token')
0077         record = A.get_authority('f')
0078         self.assertEqual((record['eic'], record['check_failed']),
0079                          (True, 'no stored GitHub token'))
0080         self.assertTrue(record['check_failed_at'] and A.may_act(record))
0081         A.set_eic('f', True, github='f-gh')
0082         record = A.get_authority('f')
0083         self.assertEqual((record['check_failed'], record['check_failed_at']),
0084                          ('', ''))
0085         with self.assertRaises(A.AuthorityError):
0086             A.set_check_failed('f', '  ')
0087
0088     def test_endpoint_takes_one_of_observation_or_failure(self):
0089         from django.test import Client
0090         client = Client(REMOTE_ADDR='127.0.0.1',
0091                         HTTP_X_REMOTE_USER=A.AUTHORITY_WRITER)
0092
0093         def post(authority):
0094             return client.post('/api/user-authority/',
0095                                {'username': 'e', 'authority': authority},
0096                                content_type='application/json')
0097
0098         self.assertEqual(post({'check_failed': 'GitHub unanswered',
0099                                'github': 'e-gh'}).status_code, 200)
0100         self.assertEqual(A.get_authority('e')['check_failed'],
0101                          'GitHub unanswered')
0102         self.assertEqual(post({'eic': True, 'check_failed': 'x'}).status_code, 400)
0103         self.assertEqual(post({'github': 'e-gh'}).status_code, 400)
0104         self.assertEqual(post({'eic': True}).status_code, 200)
0105         self.assertEqual(A.get_authority('e')['check_failed'], '')
0106
0107     def test_former_ops_rung_reads_as_the_role_and_clears_to_basic(self):
0108         A.set_rights('o', 'ops')
0109         record = A.get_authority('o')
0110         self.assertTrue(record['ops'] and A.is_ops(record) and A.may_act(record))
0111         A.set_ops('o', False)
0112         record = A.get_authority('o')
0113         self.assertEqual((record['rights'], record['ops']), ('basic', False))
0114
0115
0116 class TunnelIdentityTests(TestCase):
0117
0118     def test_no_header_means_no_identity(self):
0119         request = RequestFactory().post('/api/x/', REMOTE_ADDR='127.0.0.1',
0120                                         HTTP_AUTHORIZATION='Token nope')
0121         self.assertIsNone(TunnelAuthentication().authenticate(request))
0122         request = RequestFactory().post('/api/x/', REMOTE_ADDR='127.0.0.1',
0123                                         HTTP_X_REMOTE_USER='someone')
0124         user, _ = TunnelAuthentication().authenticate(request)
0125         self.assertEqual(user.username, 'someone')
0126
0127
0128 @override_settings(ROOT_URLCONF='swf_monitor_project.urls')
0129 class RestGateTests(TestCase):
0130
0131     def setUp(self):
0132         self.gate = AuthorityGateMiddleware(lambda request: 'passed')
0133         self.user = get_user_model().objects.create(username='person')
0134         SysConfig.update_config({A.ENFORCE_KEY: True}, username='test')
0135
0136     def _request(self, method, path, user=None, **extra):
0137         request = getattr(RequestFactory(), method)(path, **extra)
0138         request.user = user or self.user
0139         return request
0140
0141     def test_person_without_authority_is_refused_on_writes_only(self):
0142         self.assertEqual(self.gate(self._request('get', '/pcs/api/x/')),
0143                          'passed')
0144         response = self.gate(self._request('post', '/pcs/api/x/'))
0145         self.assertEqual(response.status_code, 403)
0146         self.assertIn('Join GitHub', response.content.decode())
0147
0148     def test_person_with_authority_passes(self):
0149         A.set_rights('person', 'basic')
0150         self.assertEqual(self.gate(self._request('post', '/pcs/api/x/')),
0151                          'passed')
0152
0153     def test_read_veto_outranks_membership(self):
0154         A.set_eic('person', True)
0155         A.set_rights('person', 'read')
0156         self.assertEqual(
0157             self.gate(self._request('post', '/pcs/api/x/')).status_code, 403)
0158
0159     def test_no_person_is_left_to_the_view(self):
0160         from django.contrib.auth.models import AnonymousUser
0161         request = self._request('post', '/api/logs/', user=AnonymousUser(),
0162                                 HTTP_AUTHORIZATION='Token whatever')
0163         self.assertEqual(self.gate(request), 'passed')
0164
0165     def test_exempt_writes_pass(self):
0166         request = self._request('post', '/api/user-rights/')
0167         self.assertEqual(self.gate(request), 'passed')
0168         self.assertIn('user-rights', AUTHORITY_EXEMPT_URL_NAMES)
0169
0170     def test_observing_logs_and_passes(self):
0171         SysConfig.update_config({A.ENFORCE_KEY: False}, username='test')
0172         with self.assertLogs('monitor_app.middleware', level='WARNING') as cm:
0173             self.assertEqual(self.gate(self._request('post', '/pcs/api/x/')),
0174                              'passed')
0175         self.assertIn('would refuse', cm.output[0])
0176
0177
0178 @override_settings(ROOT_URLCONF='swf_monitor_project.urls')
0179 class UserAdminCsrfTests(TestCase):
0180     """The internal face is a session with CSRF in force; the Save button
0181     must carry the token it was given."""
0182
0183     def test_rights_write_needs_and_accepts_the_csrf_token(self):
0184         from django.test import Client
0185         from django.urls import reverse
0186         staff = get_user_model().objects.create(username='admin-here',
0187                                                 is_staff=True)
0188         # A non-localhost address: the internal face. Localhost would be the
0189         # tunnel, which the middleware exempts from CSRF.
0190         client = Client(enforce_csrf_checks=True, REMOTE_ADDR='10.42.0.9')
0191         client.force_login(staff)
0192         page = client.get(reverse('monitor_app:user_admin'))
0193         self.assertEqual(page.status_code, 200)
0194         self.assertIn(b'csrfmiddlewaretoken', page.content)
0195         token = client.cookies['csrftoken'].value
0196         url = reverse('monitor_app:user-rights')
0197         body = '{"username": "someone", "rights": "basic"}'
0198         refused = client.post(url, body, content_type='application/json')
0199         self.assertEqual(refused.status_code, 403)
0200         accepted = client.post(url, body, content_type='application/json',
0201                                HTTP_X_CSRFTOKEN=token)
0202         self.assertEqual(accepted.status_code, 200, accepted.content)
0203         self.assertEqual(A.get_authority('someone')['rights'], 'basic')
0204
0205
0206 class McpDeclarationTests(TestCase):
0207
0208     def test_every_write_tool_is_guarded_and_declared(self):
0209         from monitor_app.mcp import mcp
0210         from monitor_app.mcp.common import AUTHORITY_GUARDED_TOOLS
0211         import monitor_app.mcp.ai_content  # noqa: F401  registers tools
0212         import monitor_app.mcp.ai_memory  # noqa: F401
0213         import monitor_app.mcp.system  # noqa: F401
0214         import monitor_app.mcp.workflows  # noqa: F401
0215         try:
0216             import swf_epicprod.mcp_tools.pcs  # noqa: F401
0217             import swf_epicprod.mcp_tools.proposals  # noqa: F401
0218         except ImportError:
0219             pass
0220         registered = {name: tool.fn
0221                       for name, tool in mcp._tool_manager._tools.items()}
0222         guarded = {name for name, fn in registered.items()
0223                    if getattr(fn, '__authority_guarded__', False)}
0224         # Guarded and declared are the same set: a write tool cannot be
0225         # added without declaring it, and a declared name cannot go stale.
0226         self.assertEqual(guarded, AUTHORITY_GUARDED_TOOLS & set(registered))
0227         self.assertFalse(AUTHORITY_GUARDED_TOOLS - set(registered),
0228                          'declared but not registered')