swf-remote/scripts/check_teamcomms_service_identity.py

0001 #!/usr/bin/env python3
0002 """Focused token metadata/introspection checks without database writes or HTTP."""
0003
0004 import json
0005 import os
0006 from pathlib import Path
0007 import sys
0008 from types import SimpleNamespace
0009 import unittest
0010 from unittest.mock import patch
0011
0012 sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'src'))
0013 os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'swf_remote_project.settings')
0014 import django
0015 django.setup()
0016
0017 from django.test import RequestFactory, override_settings
0018 from django.utils import timezone
0019 from remote_app.teamcomms import introspect
0020 from remote_app.token_auth import issue_token
0021
0022
0023 class IdentityChecks(unittest.TestCase):
0024     def response(self, kind='', ai=False, revoked=False, active=True):
0025         user = SimpleNamespace(pk=42, username='fixture', is_authenticated=True,
0026                                is_active=active, get_full_name=lambda: 'Fixture')
0027         token = SimpleNamespace(user=user, user_id=42, teamcomms_ai=ai,
0028                                 teamcomms_service_kind=kind, revoked=timezone.now() if revoked else None)
0029         reference = SimpleNamespace(user_id=42, token_id=7, token=token, csrf_verified=False,
0030             method='POST', path='/api/comms/messages', query_string='', body_sha256='0' * 64,
0031             expires_at=timezone.now())
0032         request = RequestFactory().post('/teamcomms-auth/introspect/',
0033             data=json.dumps({'reference': 'r' * 64}), content_type='application/json',
0034             HTTP_AUTHORIZATION='Bearer fixture-service-secret')
0035         with override_settings(SWF_TEAMCOMMS_SERVICE_TOKEN='fixture-service-secret'), \
0036              patch('remote_app.teamcomms.TeamCommsAuthReference.objects') as manager:
0037             manager.select_related.return_value.filter.return_value.first.return_value = reference
0038             response = introspect(request)
0039         return response.status_code, json.loads(response.content)
0040
0041     def test_service_subject_is_bound_to_account_not_label(self):
0042         for kind in ('program', 'connector'):
0043             with self.subTest(kind=kind):
0044                 status, body = self.response(kind)
0045                 self.assertEqual(status, 200)
0046                 self.assertEqual((body['subject'], body['account_subject'], body['kind']),
0047                                  (kind + ':42', '42', kind))
0048                 self.assertEqual(body['auth_method'], 'token')
0049                 self.assertNotIn('operator', body)
0050
0051     def test_existing_human_ai_identity_and_revocation(self):
0052         self.assertEqual(self.response()[1]['subject'], '42')
0053         ai = self.response(ai=True)[1]
0054         self.assertEqual(ai['subject'], 'ai:42')
0055         self.assertEqual(ai['operator']['subject'], '42')
0056         for kind in ('program', 'connector'):
0057             self.assertEqual(self.response(kind, revoked=True)[0], 401)
0058             self.assertEqual(self.response(kind, active=False)[0], 401)
0059
0060     def test_invalid_metadata_is_rejected_before_issuance(self):
0061         with patch('remote_app.models.ApiToken.objects.create') as create:
0062             for kind, ai in (('admin', False), ('program', True), ('connector', True)):
0063                 with self.assertRaises(ValueError):
0064                     issue_token(object(), teamcomms_ai=ai, teamcomms_service_kind=kind)
0065             create.assert_not_called()
0066
0067
0068 if __name__ == '__main__':
0069     unittest.main()