headless_api: gate request on new Repo.automation_permission (bug 1971103) r=sheehan,zeid

 * repo: add `automation_permission` and `user_can_use_automation()`
 * headless_api: gate API request to user having the correct repo's `required_automation_permission`
 * test: test headless_api gating on repo `required_automation_permission`
 * repo: fail closed on empty permissions strings
 * utils: add stub data migration command (bug 2013991)

Pull request: #1044

Author: shtrom

src/lando/conftest.py

@@ -35,6 +35,7 @@
     Worker,
 )
 from lando.main.models.landing_job import LandingJob, add_job_with_revisions
+from lando.main.models.profile import SCM_ALLOW_DIRECT_PUSH
 from lando.main.models.revision import Revision
 from lando.main.scm import SCMType
 from lando.main.scm.commit import CommitData
@@ -696,6 +697,7 @@ def git_repo_mc(
         "approval_required": approval_required,
         "autoformat_enabled": autoformat_enabled,
         "automation_enabled": automation_enabled,
+        "required_automation_permission": SCM_ALLOW_DIRECT_PUSH,
         "force_push": force_push,
         "hooks_enabled": hooks_enabled,
         "is_try": is_try,
@@ -1054,10 +1056,22 @@ def headless_permission():
 
 
 @pytest.fixture
-def headless_user(user, headless_permission):
+def direct_push_permission():
+    content_type = ContentType.objects.get_for_model(Profile)
+    perm = Permission.objects.get(
+        codename="scm_allow_direct_push", content_type=content_type
+    )
+    return perm
+
+
+@pytest.fixture
+def headless_user(
+    user, headless_permission, direct_push_permission
+) -> tuple[User, str]:
     user.user_permissions.add(headless_permission)
-    user.save()
+    user.user_permissions.add(direct_push_permission)
     user.profile.save()
+    user.save()
 
     token = ApiToken.create_token(user)
     return user, token

src/lando/headless_api/api.py

@@ -428,6 +428,14 @@ def post_repo_actions(
         )
         return 400, {"details": error}
 
+    if not repo.user_can_use_automation(request.user):
+        error = f"User {request.user.email} is not allowed to use this API for repo {repo_name}. Missing permission: {repo.required_automation_permission}."
+        logger.info(
+            error,
+            extra={"user": request.user.email, "token": request.auth.token_prefix},
+        )
+        return 403, {"details": error}
+
     with transaction.atomic():
         automation_job = AutomationJob.objects.create(
             status=JobStatus.SUBMITTED,

src/lando/headless_api/tests/test_automation_api.py

@@ -10,6 +10,8 @@
 
 import pytest
 from django.contrib.auth.hashers import check_password
+from django.contrib.auth.models import Permission, User
+from django.test.client import Client
 from pydantic import ValidationError
 
 from lando.api.legacy.workers.automation_worker import AutomationWorker
@@ -376,6 +378,65 @@ def test_automation_job_create_user_automation_disabled(
     )
 
 
+@pytest.mark.parametrize("as_superuser", (True, False))
+@pytest.mark.django_db
+def test_automation_job_create_user_no_repo_required_automation_permission(
+    client: Client,
+    headless_user: tuple[User, str],
+    make_superuser: Callable,
+    direct_push_permission: Permission,
+    repo_mc: Callable,
+    as_superuser: bool,
+):
+    user, token = headless_user
+
+    # Disable automation enabled for user.
+    user.user_permissions.remove(direct_push_permission)
+    if as_superuser:
+        user = make_superuser(user)
+    user.save()
+    user.profile.save()
+
+    repo = repo_mc(
+        scm_type=SCMType.GIT,
+        automation_enabled=True,
+    )
+
+    # Send a valid request.
+    body = {
+        "actions": [
+            {
+                "action": "add-commit",
+                "content": "0",
+                "patch_format": "git-format-patch",
+            },
+            {
+                "action": "add-commit",
+                "content": "1",
+                "patch_format": "git-format-patch",
+            },
+        ],
+    }
+    response = client.post(
+        f"/api/repo/{repo.name}",
+        data=json.dumps(body),
+        content_type="application/json",
+        headers={
+            "User-Agent": "Lando-User/testuser@example.org",
+            "Authorization": f"Bearer {token}",
+        },
+    )
+
+    assert (
+        response.status_code == 403
+    ), "User without automation permission for repo should return 403 status code."
+    response_json = response.json()
+    assert (
+        response_json["details"]
+        == f"User testuser@example.org is not allowed to use this API for repo {repo.name}. Missing permission: {repo.required_automation_permission}."
+    )
+
+
 def is_isoformat_timestamp(date_string: str) -> bool:
     """Return `True` if `date_string` is an ISO format datetime string."""
     try:

src/lando/main/migrations/0050_repo_required_automation_permission.py

@@ -0,0 +1,22 @@
+# Generated by Django 6.0.4 on 2026-04-13 07:29
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("main", "0049_profile_phabricator_phid"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="repo",
+            name="required_automation_permission",
+            field=models.CharField(
+                blank=True,
+                default="",
+                help_text="The permission required to use the automation API against this repo. For example, main.scm_allow_direct_push.",
+            ),
+        ),
+    ]

src/lando/main/models/repo.py

@@ -212,6 +212,11 @@ def path(self) -> str:
 
     # Use this field to enable/disable access to this repo via the automation API.
     automation_enabled = models.BooleanField(default=False)
+    required_automation_permission = models.CharField(
+        help_text="The permission required to use the automation API against this repo. For example, main.scm_allow_direct_push.",
+        blank=True,
+        default="",
+    )
 
     # Use this field to enable/disable access to this repo via the try API.
     is_try = models.BooleanField(default=False)
@@ -395,28 +400,49 @@ def phab_identifier(self) -> str | None:
 
     def user_can_push(self, user: User) -> bool:
         """
-        Test that the user has permission to push to this repo.
+        Test that the user has the permission to push to this repo.
+        """
+        return self._user_has_direct_permission(user, str(self.required_permission))
+
+    def user_can_use_automation(self, user: User) -> bool:
+        """
+        Test that the user has the permission to create automation job for this repo.
+        """
+        return self._user_has_direct_permission(
+            user, str(self.required_automation_permission)
+        )
+
+    def _user_has_direct_permission(self, user: User, permission: str):
+        """
+        Test that the user has a specific permission directly rather than via a role.
 
         If the user is a superuser, this checks that the user was given the permissions
         directly, rather than transitively by virtue of being an admin.
 
         Parameters:
 
         user: User
-            User to check permmission.
+            User for whom to check permission.
+
+        permissions: str
+            Permission string to look for. If empty, fail closed denying any access.
 
         Returns:
             bool: whether the user has the permission
         """
+        # Fail closed if the permission is empty.
+        if not permission:
+            return False
+
         if user.is_superuser:
             # We can't rely on the `get_user_permissions()` method, as it returns all existing
             # permissions for superusers. Here, we want to check permissions that have been
             # explicitly given to the user from LDAP groups.
 
-            app_label, codename = self.required_permission.split(".", maxsplit=1)
+            app_label, codename = permission.split(".", maxsplit=1)
             return user.user_permissions.filter(
                 content_type__app_label=app_label, codename=codename
             ).exists()
 
         # If the user is not a superuser, we can skip the DB round-trip.
-        return self.required_permission in user.get_user_permissions()
+        return permission in user.get_user_permissions()

src/lando/main/tests/test_models.py

@@ -5,6 +5,7 @@
 
 import pytest
 from django.conf import settings
+from django.contrib.auth.models import Permission, User
 from django.core.exceptions import ValidationError
 
 from lando.main.models import CommitMap, Repo
@@ -191,6 +192,42 @@ def test__models__Repo__git_repo_name(
         assert repo.git_repo_name == expected_git_repo_name
 
 
+@pytest.mark.parametrize(
+    "has_permission,as_superuser",
+    (
+        (False, False),
+        (False, True),
+        (True, False),
+        (True, True),
+    ),
+)
+@pytest.mark.django_db
+def test__models__Repo___user_has_direct_permissions(
+    user: User,
+    make_superuser: Callable,
+    repo_mc: Callable,
+    direct_push_permission: Permission,
+    has_permission: bool,
+    as_superuser: bool,
+):
+    permission = direct_push_permission
+    permission_string = f"{permission.content_type.app_label}.{permission.codename}"
+
+    if has_permission:
+        user.user_permissions.add(permission)
+    if as_superuser:
+        user = make_superuser(user)
+    user.save()
+    user.profile.save()
+
+    repo = repo_mc(
+        scm_type=SCMType.GIT,
+        automation_enabled=True,
+    )
+
+    assert repo._user_has_direct_permission(user, permission_string) == has_permission
+
+
 @pytest.mark.parametrize(
     "pulse_routing_key,expected_valid",
     (

src/lando/utils/management/commands/migrate_data.py

@@ -0,0 +1,99 @@
+import argparse
+
+from django.core.management import BaseCommand, CommandError, CommandParser
+
+from lando.main.models import SCM_ALLOW_DIRECT_PUSH
+from lando.main.models.repo import Repo
+
+
+class Command(BaseCommand):
+    help = """This stub commands allows to run data migrations.
+
+    It should be fleshed out as part of bug 2013991.
+    """
+    # TODO: record already-applied migrations.
+
+    def add_arguments(self, parser: CommandParser):
+        parser.add_argument(
+            "-l", "--list", default=False, action=argparse.BooleanOptionalAction
+        )
+        parser.add_argument(
+            "-y",
+            "--yes",
+            default=False,
+            help="Don't ask for confirmation",
+            action=argparse.BooleanOptionalAction,
+        )
+        parser.add_argument("migration", help="Migration to apply", nargs="?")
+
+    def handle(self, *args, **options):
+        ask_confirm = not options.get("yes", False)
+
+        if options.get("list"):
+            self.stdout.write("Available data migrations:\n")
+            for migration in [
+                method.removeprefix("migrate_")
+                for method in dir(self)
+                if method.startswith("migrate_")
+            ]:
+                self.stdout.write(f"* {migration}")
+            raise SystemExit()
+
+        if not (migration := options.get("migration")):
+            raise CommandError("No migration specified")
+
+        method_name = f"migrate_{migration}"
+
+        if not (migration_method := getattr(self, method_name)):
+            raise CommandError("Migration method not found: {method_name}")
+
+        migration_method(ask_confirm)
+
+    def _get_confirmation(self, ask_confirm: bool, prompt: str, details: str):
+        """Show a confirmation prompt with details.
+
+        If ask_confirm is set, details are printed prior to requesting confirmation.
+
+        Otherwise, let the caller show the details as it proceeds.
+        """
+        self.stdout.write(prompt, ending="")
+        if ask_confirm:
+            self.stdout.write(f"{details}. Confirm [yN]? ", ending="")
+            if input().lower() != "y":
+                raise CommandError("User interruption")
+
+            self.stdout.write("Progress: ", ending="")
+
+    #
+    # MIGRATION METHODS
+    #
+
+    def migrate_1_bug1971103_add_firefox_required_automation_permissions(
+        self, ask_confirm: bool = True
+    ):
+        """
+        Explicitly set the  `required_automation_permission` for all Firefox repos.
+
+        This is to match the currectly implicit authorisation for Firefox sheriffs.
+        """
+        fx_repos = Repo.objects.filter(
+            name__startswith="firefox-", required_automation_permission=""
+        )
+        if not fx_repos:
+            self.stdout.write(
+                "No firefox-% repo found with NULL required_automation_permission."
+            )
+            raise SystemExit()
+
+        fx_repo_names = ", ".join(repo.name for repo in fx_repos)
+        self._get_confirmation(
+            ask_confirm, "Updating required_automation_permission for: ", fx_repo_names
+        )
+
+        for repo in fx_repos:
+            if not repo.required_automation_permission:
+                repo.required_automation_permission = SCM_ALLOW_DIRECT_PUSH
+                repo.save()
+                self.stdout.write(f"{repo.name} ", ending="")
+
+        self.stdout.write("done.")