diff --git a/ngen/management/commands/syncgroups.py b/ngen/management/commands/syncgroups.py new file mode 100644 index 00000000..86f3f6ab --- /dev/null +++ b/ngen/management/commands/syncgroups.py @@ -0,0 +1,218 @@ +import json +import os + +from django.apps import apps +from django.contrib.auth.models import Group, Permission +from django.core.management.base import BaseCommand, CommandError +from django.db import transaction + +WRITE_ACTIONS = ("add", "change", "delete") + + +class Command(BaseCommand): + help = ( + "Reconciles the groups and their permissions with the ones defined in " + "ngen/fixtures/group.json. The fixture is only loaded on a brand new " + "installation, so the permissions a release adds to a role never reach " + "the installations already running. Unlike 'loaddata group' this is " + "additive: it grants what is missing and does not touch anything else, " + "so the permissions an administrator added by hand are kept." + ) + + def add_arguments(self, parser): + parser.add_argument( + "--fixture", + help="Path of the fixture to read (defaults to the one shipped with ngen)", + ) + parser.add_argument( + "--group", + action="append", + dest="groups", + help="Only this group, can be repeated", + ) + parser.add_argument( + "--dry-run", + action="store_true", + help="Print what would change without applying it", + ) + parser.add_argument( + "--check", + action="store_true", + help="Exit with 1 if there is anything to apply, implies --dry-run", + ) + parser.add_argument( + "--prune", + action="store_true", + help="Also revoke the permissions the fixture does not define", + ) + + def handle(self, *args, **options): + dry_run = options["dry_run"] or options["check"] + groups_filter = options["groups"] + + entries = self.read_fixture(options["fixture"]) + if groups_filter: + entries = [e for e in entries if e["name"] in groups_filter] + missing = set(groups_filter) - {e["name"] for e in entries} + if missing: + raise CommandError( + f"The fixture has no group named {', '.join(sorted(missing))}" + ) + + pending = 0 + for entry in entries: + pending += self.sync_group(entry, dry_run=dry_run, prune=options["prune"]) + + if dry_run and pending: + # Reporting now would list problems the pending changes may fix + self.stdout.write( + "Apply the changes to see which permissions stay unusable." + ) + else: + self.report_unusable_permissions([entry["name"] for entry in entries]) + + if not pending: + self.stdout.write(self.style.SUCCESS("Every group is already in sync.")) + elif dry_run: + self.stdout.write( + self.style.WARNING(f"{pending} change(s) pending, nothing was applied.") + ) + else: + self.stdout.write(self.style.SUCCESS(f"{pending} change(s) applied.")) + + if options["check"] and pending: + raise SystemExit(1) + + def read_fixture(self, path): + """ + Groups of the fixture as {name, permissions}, resolving the permissions + to the ones existing in the database + """ + path = path or os.path.join( + apps.get_app_config("ngen").path, "fixtures", "group.json" + ) + try: + with open(path, encoding="utf-8") as fixture: + data = json.load(fixture) + except OSError as error: + raise CommandError(f"Could not read {path}: {error}") from error + + # Every permission at once: the shipped fixture holds hundreds of them + # and resolving one by one is one query each + known = { + ( + permission.codename, + permission.content_type.app_label, + permission.content_type.model, + ): permission + for permission in Permission.objects.select_related("content_type") + } + + entries = [] + for obj in data: + if obj.get("model") != "auth.group": + continue + permissions, unknown = self.resolve_permissions( + obj["fields"].get("permissions", []), known + ) + for codename in unknown: + self.stdout.write( + self.style.WARNING( + f" the fixture asks for '{codename}', which does not exist " + "in this installation" + ) + ) + entries.append({"name": obj["fields"]["name"], "permissions": permissions}) + return entries + + @staticmethod + def resolve_permissions(natural_keys, known): + permissions, unknown = [], [] + for codename, app_label, model in natural_keys: + permission = known.get((codename, app_label, model)) + if permission: + permissions.append(permission) + else: + unknown.append(codename) + return permissions, unknown + + def sync_group(self, entry, dry_run, prune): + """ + Grant the permissions of the fixture the group does not have, and revoke + the ones it has beyond it only when pruning. Returns how many changed. + """ + group, created = ( + Group.objects.get_or_create(name=entry["name"]) + if not dry_run + else (Group.objects.filter(name=entry["name"]).first(), False) + ) + if group is None: + self.stdout.write(self.style.MIGRATE_HEADING(entry["name"])) + self.stdout.write( + f" + would be created with {len(entry['permissions'])} permission(s)" + ) + # The group itself counts as a change, or a group defined with no + # permissions would be reported as nothing to do + return 1 + len(entry["permissions"]) + + current = set(group.permissions.all()) + expected = set(entry["permissions"]) + to_add = sorted(expected - current, key=lambda p: p.codename) + to_remove = ( + sorted(current - expected, key=lambda p: p.codename) if prune else [] + ) + + if not (created or to_add or to_remove): + return 0 + + self.stdout.write(self.style.MIGRATE_HEADING(entry["name"])) + if created: + self.stdout.write(" + created") + for permission in to_add: + self.stdout.write(self.style.SUCCESS(f" + {permission.codename}")) + for permission in to_remove: + self.stdout.write(self.style.ERROR(f" - {permission.codename}")) + + if not dry_run: + with transaction.atomic(): + group.permissions.add(*to_add) + if to_remove: + group.permissions.remove(*to_remove) + + return (1 if created else 0) + len(to_add) + len(to_remove) + + def report_unusable_permissions(self, group_names): + """ + DRF maps GET to view_, so a group that can add, change or delete a + model without being able to view it cannot use those permissions through + the API. It is worth saying, it is how the Incident Responder role ended + up unable to read the playbooks it was meant to run. + """ + for name in group_names: + group = Group.objects.filter(name=name).first() + if not group: + continue + + actions = {} + # Only the models of ngen: the ones of the packages it depends on are + # not served by its api, so the mapping does not apply to them + for permission in group.permissions.filter( + content_type__app_label="ngen" + ).select_related("content_type"): + action, _, model = permission.codename.partition("_") + if action in WRITE_ACTIONS + ("view",) and model: + actions.setdefault(model, set()).add(action) + + unusable = sorted( + model + for model, done in actions.items() + if "view" not in done and done & set(WRITE_ACTIONS) + ) + if unusable: + self.stdout.write( + self.style.WARNING( + f"{name}: can write but not read {len(unusable)} model(s), " + "so those permissions cannot be used through the api: " + f"{', '.join(unusable)}" + ) + ) diff --git a/ngen/tests/commands/__init__.py b/ngen/tests/commands/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/ngen/tests/commands/test_syncgroups.py b/ngen/tests/commands/test_syncgroups.py new file mode 100644 index 00000000..9ab88057 --- /dev/null +++ b/ngen/tests/commands/test_syncgroups.py @@ -0,0 +1,239 @@ +""" +Django syncgroups management command tests. +""" + +import json +import tempfile +from io import StringIO +from pathlib import Path + +from django.contrib.auth.models import Group, Permission +from django.core.management import CommandError, call_command +from django.test import TestCase + + +class SyncGroupsCommandTestCase(TestCase): + """ + This will handle the syncgroups command testcases + """ + + @classmethod + def setUpTestData(cls): + cls.view_playbook = Permission.objects.get( + codename="view_playbook", content_type__app_label="ngen" + ) + cls.change_playbook = Permission.objects.get( + codename="change_playbook", content_type__app_label="ngen" + ) + cls.view_task = Permission.objects.get( + codename="view_task", content_type__app_label="ngen" + ) + + def setUp(self): + self.tempdir = tempfile.TemporaryDirectory() + self.addCleanup(self.tempdir.cleanup) + + def write_fixture(self, groups): + """ + Fixture with the same shape as ngen/fixtures/group.json + """ + path = Path(self.tempdir.name) / "group.json" + path.write_text( + json.dumps( + [ + { + "model": "auth.group", + "pk": position, + "fields": { + "name": name, + "permissions": [ + [ + permission.codename, + "ngen", + permission.content_type.model, + ] + for permission in permissions + ], + }, + } + for position, (name, permissions) in enumerate( + groups.items(), start=1 + ) + ] + ), + encoding="utf-8", + ) + return str(path) + + def call(self, fixture, *args): + out = StringIO() + call_command("syncgroups", "--fixture", fixture, *args, stdout=out) + return out.getvalue() + + def test_grants_the_missing_permissions(self): + """ + Test that the permissions of the fixture the group does not have are granted + """ + group = Group.objects.create(name="Responder") + group.permissions.add(self.change_playbook) + fixture = self.write_fixture( + {"Responder": [self.change_playbook, self.view_playbook]} + ) + + output = self.call(fixture) + + self.assertIn("view_playbook", output) + self.assertQuerysetEqual( + group.permissions.all(), + [self.change_playbook, self.view_playbook], + ordered=False, + ) + + def test_keeps_the_permissions_added_by_the_administrator(self): + """ + Test that it does not revoke anything, which is what tells it apart from + loaddata: a permission granted by hand survives + """ + group = Group.objects.create(name="Responder") + group.permissions.add(self.change_playbook, self.view_task) + fixture = self.write_fixture({"Responder": [self.change_playbook]}) + + self.call(fixture) + + self.assertIn(self.view_task, group.permissions.all()) + + def test_prune_revokes_what_the_fixture_does_not_define(self): + """ + Test that pruning does revoke them, which is opt in + """ + group = Group.objects.create(name="Responder") + group.permissions.add(self.change_playbook, self.view_task) + fixture = self.write_fixture({"Responder": [self.change_playbook]}) + + output = self.call(fixture, "--prune") + + self.assertIn("- view_task", output) + self.assertQuerysetEqual(group.permissions.all(), [self.change_playbook]) + + def test_dry_run_changes_nothing(self): + """ + Test that a dry run only reports + """ + group = Group.objects.create(name="Responder") + fixture = self.write_fixture({"Responder": [self.view_playbook]}) + + output = self.call(fixture, "--dry-run") + + self.assertIn("view_playbook", output) + self.assertIn("nothing was applied", output) + self.assertEqual(group.permissions.count(), 0) + + def test_check_exits_with_an_error_when_there_is_something_to_apply(self): + """ + Test the mode meant for CI + """ + group = Group.objects.create(name="Responder") + fixture = self.write_fixture({"Responder": [self.view_playbook]}) + + with self.assertRaises(SystemExit): + self.call(fixture, "--check") + + self.assertEqual(group.permissions.count(), 0) + + group.permissions.add(self.view_playbook) + self.call(fixture, "--check") + + def test_creates_a_group_that_does_not_exist(self): + """ + Test that a group added by a release is created + """ + fixture = self.write_fixture({"Brand new": [self.view_playbook]}) + + self.call(fixture) + + self.assertQuerysetEqual( + Group.objects.get(name="Brand new").permissions.all(), [self.view_playbook] + ) + + def test_creating_a_group_counts_as_a_change(self): + """ + Test that a group of the fixture that does not exist is pending even + when it defines no permissions, which --check has to catch + """ + fixture = self.write_fixture({"Empty": []}) + + with self.assertRaises(SystemExit): + self.call(fixture, "--check") + + output = self.call(fixture) + + self.assertNotIn("already in sync", output) + self.assertTrue(Group.objects.filter(name="Empty").exists()) + + def test_does_not_report_what_the_pending_changes_would_fix(self): + """ + Test that a dry run with something to apply does not list the + permissions that applying it may make usable + """ + Group.objects.create(name="Responder").permissions.add(self.change_playbook) + fixture = self.write_fixture( + {"Responder": [self.change_playbook, self.view_playbook]} + ) + + output = self.call(fixture, "--dry-run") + + self.assertNotIn("can write but not read", output) + self.assertIn("Apply the changes", output) + + def test_only_the_given_group(self): + """ + Test that --group limits the reconciliation + """ + first = Group.objects.create(name="Responder") + second = Group.objects.create(name="Analist") + fixture = self.write_fixture( + {"Responder": [self.view_playbook], "Analist": [self.view_task]} + ) + + self.call(fixture, "--group", "Responder") + + self.assertEqual(first.permissions.count(), 1) + self.assertEqual(second.permissions.count(), 0) + + def test_unknown_group_is_an_error(self): + """ + Test that asking for a group the fixture does not define fails + """ + fixture = self.write_fixture({"Responder": [self.view_playbook]}) + + with self.assertRaises(CommandError): + self.call(fixture, "--group", "Does not exist") + + def test_reports_the_permissions_that_cannot_be_used(self): + """ + Test the report of write permissions without their view counterpart, + which DRF needs to answer a GET + """ + group = Group.objects.create(name="Responder") + group.permissions.add(self.change_playbook) + fixture = self.write_fixture({"Responder": [self.change_playbook]}) + + output = self.call(fixture) + + self.assertIn("can write but not read", output) + self.assertIn("playbook", output) + + def test_reports_nothing_when_every_write_has_its_view(self): + """ + Test that the report stays quiet when there is nothing to say + """ + group = Group.objects.create(name="Responder") + group.permissions.add(self.change_playbook, self.view_playbook) + fixture = self.write_fixture( + {"Responder": [self.change_playbook, self.view_playbook]} + ) + + output = self.call(fixture) + + self.assertNotIn("can write but not read", output) + self.assertIn("already in sync", output)