diff --git a/CHANGES/8101.feature b/CHANGES/8101.feature
new file mode 100644
index 000000000..7dff74a71
--- /dev/null
+++ b/CHANGES/8101.feature
@@ -0,0 +1 @@
+Added Owner, Collaborator, and Consumer groups and permissions for Namespaces and Repositories.
diff --git a/pulp_container/app/access_policy.py b/pulp_container/app/access_policy.py
index 4f88bc787..5410ece08 100644
--- a/pulp_container/app/access_policy.py
+++ b/pulp_container/app/access_policy.py
@@ -6,46 +6,37 @@
 from pulp_container.app import models
 
 
-class NamespacePermissionsChecker:
+class NamespaceAccessPolicyMixin:
     """
-    A class that contains a function which checks permissions required for modifying namespaces.
+    Access policy mixin for ContainerDistributionViewSet and ContainerPushRepositoryViewSet which
+    handles namespace permissions.
     """
 
-    @staticmethod
-    def has_permissions(namespace, user, permission):
+    def has_namespace_obj_perms(self, request, view, action, permission):
         """
-        Check whether a user have permissions to manage the passed namespace.
+        Check if a user has object-level perms on the namespace associated with the distribution
+        or repository.
         """
-
-        try:
-            namespace = models.ContainerNamespace.objects.get(name=namespace)
-        except models.ContainerNamespace.DoesNotExist:
-            # check model permissions for namespace creation
-            return user.has_perm("container.add_containernamespace")
+        obj = view.get_object()
+        if type(obj) == models.ContainerDistribution:
+            namespace = obj.namespace
+            return request.user.has_perm(permission, namespace)
         else:
-            # existing namespace
-            return user.has_perm(permission) or user.has_perm(permission, namespace)
+            dists_qs = models.ContainerDistribution.objects.filter(repository=obj)
+            for dist in dists_qs:
+                if request.user.has_perm(permission, dist.namespace):
+                    return True
+        return False
 
-
-class DistributionAccessPolicyMixin:
-    """
-    Access policy mixin for DistributionViewSet which handles namespace permissions.
-    """
-
-    def has_manage_namespace_dist_perms(self, request, view, action, permission):
+    def has_namespace_or_obj_perms(self, request, view, action, permission):
         """
-        Check whether a user can create a namespace or it can manage distributions
-        for existing namespace.
+        Check if a user has a namespace-level perms or object-level permission
         """
-        namespace = request.data["base_path"].split("/")[0]
-        return NamespacePermissionsChecker.has_permissions(namespace, request.user, permission)
-
-    def has_namespace_obj_perms(self, request, view, action, permission):
-        """
-        Check if a user has object-level perms on the namespace associated with the distribution.
-        """
-        namespace = view.get_object().namespace
-        return request.user.has_perm(permission, namespace)
+        ns_perm = "container.namespace_{}".format(permission.split(".", 1)[1])
+        if self.has_namespace_obj_perms(request, view, action, ns_perm):
+            return True
+        else:
+            return request.user.has_perm(permission, view.get_object())
 
     def obj_exists(self, request, view, action):
         """
@@ -60,20 +51,22 @@ def is_private(self, request, view, action):
         return view.get_object().private
 
 
-class DistributionAccessPolicyFromDB(AccessPolicyFromDB, DistributionAccessPolicyMixin):
+class NamespaceAccessPolicyFromDB(AccessPolicyFromDB, NamespaceAccessPolicyMixin):
     """
-    Access policy for DistributionViewSet which handles namespace permissions.
+    Access policy for ContainerDistributionViewSet and ContainerPushRepositoryViewSet which handles
+    namespace permissions.
     """
 
 
-class RegistryAccessPolicy(AccessPolicy, DistributionAccessPolicyMixin):
+class RegistryAccessPolicy(AccessPolicy, NamespaceAccessPolicyMixin):
     """
-    An AccessPolicy that loads statements from the container distribution viewset.
+    An AccessPolicy that loads statements from the ContainerDistribution, ContainerNamespace,
+    and ContainerPushRepository viewsets.
     """
 
     def get_policy_statements(self, request, view):
         """
-        Return the policy statements for the container distribution viewset.
+        Return the policy statements for the container distribution and namespace viewsets.
 
         Args:
             request (rest_framework.request.Request): The request being checked for authorization.
@@ -83,8 +76,12 @@ def get_policy_statements(self, request, view):
             The access policy statements in drf-access-policy policy structure.
 
         """
-
-        access_policy_obj = AccessPolicyModel.objects.get(
-            viewset_name="distributions/container/container"
-        )
+        if isinstance(view.get_object(), models.ContainerDistribution):
+            access_policy_obj = AccessPolicyModel.objects.get(
+                viewset_name="distributions/container/container"
+            )
+        else:
+            access_policy_obj = AccessPolicyModel.objects.get(
+                viewset_name="pulp_container/namespaces"
+            )
         return access_policy_obj.statements
diff --git a/pulp_container/app/authorization.py b/pulp_container/app/authorization.py
index 69327c729..d6790e54a 100644
--- a/pulp_container/app/authorization.py
+++ b/pulp_container/app/authorization.py
@@ -15,7 +15,7 @@
 from cryptography.hazmat.backends import default_backend
 from cryptography.hazmat.primitives import serialization
 
-from pulp_container.app.models import ContainerDistribution
+from pulp_container.app.models import ContainerDistribution, ContainerNamespace
 from pulp_container.app.access_policy import RegistryAccessPolicy
 
 TOKEN_EXPIRATION_TIME = settings.get("TOKEN_EXPIRATION_TIME", 300)
@@ -151,6 +151,18 @@ def determine_access(self):
 
         return {"type": typ, "name": name, "actions": list(permitted_actions)}
 
+    def has_permission(self, obj, method, action, data):
+        """Check if user has permission to perform action."""
+
+        # Fake the request
+        request = Request(HttpRequest())
+        request.method = method
+        request.user = self.user
+        request._full_data = data
+        # Fake the corresponding view
+        view = namedtuple("FakeView", ["action", "get_object"])(action, lambda: obj)
+        return self.access_policy.has_permission(request, view)
+
     def has_pull_permissions(self, path):
         """
         Check if the user has permissions to pull from the repository specified by the path.
@@ -158,16 +170,18 @@ def has_pull_permissions(self, path):
         try:
             distribution = ContainerDistribution.objects.get(base_path=path)
         except ContainerDistribution.DoesNotExist:
-            return False
+            namespace_name = path.split("/")[0]
+            try:
+                namespace = ContainerNamespace.objects.get(name=namespace_name)
+            except ContainerNamespace.DoesNotExist:
+                # Check if user is allowed to create a new namespace
+                return self.has_permission(None, "POST", "create", {"name": namespace_name})
+            # Check if user is allowed to view distributions in the namespace
+            return self.has_permission(
+                namespace, "GET", "view_distribution", {"name": namespace_name}
+            )
 
-        # Fake the request
-        request = Request(HttpRequest())
-        request.method = "GET"
-        request.user = self.user
-        request._full_data = {"base_path": path}
-        # Fake the corresponding view
-        view = namedtuple("FakeView", ["action", "get_object"])("pull", lambda: distribution)
-        return self.access_policy.has_permission(request, view)
+        return self.has_permission(distribution, "GET", "pull", {"base_path": path})
 
     def has_push_permissions(self, path):
         """
@@ -176,16 +190,16 @@ def has_push_permissions(self, path):
         try:
             distribution = ContainerDistribution.objects.get(base_path=path)
         except ContainerDistribution.DoesNotExist:
-            distribution = None
-
-        # Fake the request
-        request = Request(HttpRequest())
-        request.method = "POST"
-        request.user = self.user
-        request._full_data = {"base_path": path}
-        # Fake the corresponding view
-        view = namedtuple("FakeView", ["action", "get_object"])("push", lambda: distribution)
-        return self.access_policy.has_permission(request, view)
+            namespace_name = path.split("/")[0]
+            try:
+                namespace = ContainerNamespace.objects.get(name=namespace_name)
+            except ContainerNamespace.DoesNotExist:
+                # Check if user is allowed to create a new namespace
+                return self.has_permission(None, "POST", "create", {"name": namespace_name})
+            # Check if user is allowed to create a new distribution in the namespace
+            return self.has_permission(namespace, "POST", "create_distribution", {})
+
+        return self.has_permission(distribution, "POST", "push", {"base_path": path})
 
     def has_view_catalog_permissions(self, path):
         """
diff --git a/pulp_container/app/migrations/0017_add_granular_perms.py b/pulp_container/app/migrations/0017_add_granular_perms.py
new file mode 100644
index 000000000..f29d237ea
--- /dev/null
+++ b/pulp_container/app/migrations/0017_add_granular_perms.py
@@ -0,0 +1,17 @@
+# Generated by Django 2.2.17 on 2021-02-04 02:18
+
+from django.db import migrations
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('container', '0016_add_delete_versions_permission'),
+    ]
+
+    operations = [
+        migrations.AlterModelOptions(
+            name='containernamespace',
+            options={'permissions': [('namespace_add_containerdistribution', 'Add any distribution to a namespace'), ('namespace_delete_containerdistribution', 'Delete any distribution from a namespace'), ('namespace_view_containerdistribution', 'View any distribution in a namespace'), ('namespace_pull_containerdistribution', 'Pull from any distribution in a namespace'), ('namespace_push_containerdistribution', 'Push to any distribution in a namespace'), ('namespace_change_containerdistribution', 'Change any distribution in a namespace'), ('namespace_view_containerpushrepository', 'View any push repository in a namespace'), ('namespace_modify_content_containerpushrepository', 'Modify content in any push repository in a namespace')]},
+        ),
+    ]
diff --git a/pulp_container/app/models.py b/pulp_container/app/models.py
index 828c53e8f..4b276d168 100644
--- a/pulp_container/app/models.py
+++ b/pulp_container/app/models.py
@@ -9,9 +9,16 @@
 from urllib.parse import urlparse
 
 from django.db import models
+from django.contrib.auth.models import Group
 from django.contrib.postgres import fields
 from django.shortcuts import redirect
 
+from django_currentuser.middleware import get_current_authenticated_user
+from django_lifecycle import hook
+
+from guardian.models.models import GroupObjectPermission, UserObjectPermission
+from guardian.shortcuts import assign_perm
+
 from pulpcore.plugin.download import DownloaderFactory
 from pulpcore.plugin.models import (
     AutoAddObjPermsMixin,
@@ -188,7 +195,7 @@ class Meta:
         unique_together = (("name", "tagged_manifest"),)
 
 
-class ContainerNamespace(BaseModel, AutoAddObjPermsMixin, AutoDeleteObjPermsMixin):
+class ContainerNamespace(BaseModel, AutoAddObjPermsMixin):
     """
     Namespace for the container registry.
     """
@@ -196,10 +203,59 @@ class ContainerNamespace(BaseModel, AutoAddObjPermsMixin, AutoDeleteObjPermsMixi
     name = models.CharField(max_length=255, db_index=True)
     ACCESS_POLICY_VIEWSET_NAME = "pulp_container/namespaces"
 
+    def create_namespace_group(self, permissions, parameters):
+        """
+        Creates a namespace group and optionally adds the current user to it.
+
+        The parameters are specified as a dictionary with the following keys:
+
+        "group_type" - the type of group - owners, collaborators, or consumers
+        "add_user_to_group" - a boolean that specifies if the current user should be added to the
+                              group.
+
+        The permissions are object level permissions assigned to the group.
+        """
+
+        group_type = parameters["group_type"]
+        add_user_to_group = parameters["add_user_to_group"]
+        group = Group.objects.create(
+            name="{}.{}.{}".format("container.namespace", group_type, self.name)
+        )
+        current_user = get_current_authenticated_user()
+        owners_group = Group.objects.get(
+            name="{}.{}".format("container.namespace.owners", self.name)
+        )
+        assign_perm("auth.change_group", owners_group, group)
+        assign_perm("auth.view_group", owners_group, group)
+        if add_user_to_group:
+            current_user.groups.add(group)
+        self.add_for_groups(permissions, group.name)
+
+    @hook("before_delete")
+    def delete_groups_and_user_obj_perms(self):
+        """
+        Delete all auto created groups associated with this Namespace and user object
+        permissions.
+        """
+        group_name_regex = r"container.namespace.(.*).{}".format(self.name)
+        Group.objects.filter(name__regex=group_name_regex).delete()
+        UserObjectPermission.objects.filter(object_pk=self.pk).delete()
+        GroupObjectPermission.objects.filter(object_pk=self.pk).delete()
+
     class Meta:
         unique_together = (("name",),)
         permissions = [
-            ("manage_namespace_distributions", "Can manage distributions in a namespace"),
+            ("namespace_add_containerdistribution", "Add any distribution to a namespace"),
+            ("namespace_delete_containerdistribution", "Delete any distribution from a namespace"),
+            ("namespace_view_containerdistribution", "View any distribution in a namespace"),
+            ("namespace_pull_containerdistribution", "Pull from any distribution in a namespace"),
+            ("namespace_push_containerdistribution", "Push to any distribution in a namespace"),
+            ("namespace_change_containerdistribution", "Change any distribution in a namespace"),
+            ("namespace_view_containerpushrepository", "View any push repository in a namespace"),
+            (
+                "namespace_modify_content_containerpushrepository",
+                "Modify content in any push repository in a namespace",
+            ),
         ]
 
 
@@ -255,6 +311,34 @@ class ContainerPushRepository(Repository, AutoAddObjPermsMixin, AutoDeleteObjPer
     PUSH_ENABLED = True
     ACCESS_POLICY_VIEWSET_NAME = "repositories/container/container-push"
 
+    def add_perms_to_distribution_group(self, permissions, parameters):
+        """
+        Adds push repository object permissions to a distribution group.
+
+        The parameters are specified as a dictionary with the following keys:
+
+        "group_type" - the type of group - owners, collaborators, or consumers
+        "add_user_to_group" - a boolean that specifies if the current user should be added to the
+                              group.
+
+        The permissions are object level permissions assigned to the group.
+        """
+
+        group_type = parameters["group_type"]
+        add_user_to_group = parameters["add_user_to_group"]
+        try:
+            suffix = ContainerDistribution.objects.get(repository=self).pk
+        except ContainerDistribution.DoesNotExist:
+            # The distribution has not been created yet
+            return
+        group = Group.objects.get(
+            name="{}.{}.{}".format("container.distribution", group_type, suffix)
+        )
+        current_user = get_current_authenticated_user()
+        if add_user_to_group:
+            current_user.groups.add(group)
+        self.add_for_groups(permissions, group.name)
+
     class Meta:
         default_related_name = "%(app_label)s_%(model_name)s"
         permissions = [
@@ -361,9 +445,7 @@ class Meta:
         default_related_name = "%(app_label)s_%(model_name)s"
 
 
-class ContainerDistribution(
-    RepositoryVersionDistribution, AutoAddObjPermsMixin, AutoDeleteObjPermsMixin
-):
+class ContainerDistribution(RepositoryVersionDistribution, AutoAddObjPermsMixin):
     """
     A container distribution defines how a repository version is distributed by Pulp's webserver.
     """
@@ -404,6 +486,65 @@ def redirect_to_content_app(self, url):
             url = self.content_guard.cast().preauthenticate_url(url)
         return redirect(url)
 
+    def create_distribution_group(self, permissions, parameters):
+        """
+        Creates a distribution group and optionally adds the current user to it.
+
+        The parameters are specified as a dictionary with the following keys:
+
+        "group_type" - the type of group - owners, collaborators, or consumers
+        "add_user_to_group" - a boolean that specifies if the current user should be added to the
+                              group.the "model_field" for the instance.
+
+        The permissions are object level permissions assigned to the group.
+        """
+
+        group_type = parameters["group_type"]
+        add_user_to_group = parameters["add_user_to_group"]
+
+        group = Group.objects.create(
+            name="{}.{}.{}".format("container.distribution", group_type, self.pk)
+        )
+        current_user = get_current_authenticated_user()
+        owners_group = Group.objects.get(
+            name="{}.{}".format("container.distribution.owners", self.pk)
+        )
+        assign_perm("auth.change_group", owners_group, group)
+        assign_perm("auth.view_group", owners_group, group)
+        if add_user_to_group:
+            current_user.groups.add(group)
+        self.add_for_groups(permissions, group.name)
+
+    def add_push_repository_perms_to_distribution_group(self, permissions, parameters):
+        """
+        Adds permissions related to ContainerPushRepository to a distribution group.
+
+        The parameters are specified as a dictionary with the following keys:
+
+        "group_type" - the type of group - owners, collaborators, or consumers
+
+        The permissions are ContainerPushRepository object level permissions assigned to the
+        group. The repository is the one that is associated with the ContainerDistribution.
+        """
+
+        group_type = parameters["group_type"]
+        group = Group.objects.get(
+            name="{}.{}.{}".format("container.distribution", group_type, self.pk)
+        )
+        if isinstance(self.repository, ContainerPushRepository):
+            self.repository.add_for_groups(permissions, group.name)
+
+    @hook("before_delete")
+    def delete_groups_and_user_obj_perms(self):
+        """
+        Delete all auto created groups associated with this Distribution and user object
+        permissions.
+        """
+        group_name_regex = r"container.distribution.(.*).{}".format(self.pk)
+        Group.objects.filter(name__regex=group_name_regex).delete()
+        UserObjectPermission.objects.filter(object_pk=self.pk).delete()
+        GroupObjectPermission.objects.filter(object_pk=self.pk).delete()
+
     class Meta:
         default_related_name = "%(app_label)s_%(model_name)s"
         permissions = [
diff --git a/pulp_container/app/viewsets.py b/pulp_container/app/viewsets.py
index 63ada0bae..a4882f7dd 100644
--- a/pulp_container/app/viewsets.py
+++ b/pulp_container/app/viewsets.py
@@ -9,8 +9,10 @@
 from gettext import gettext as _
 
 from django.db import IntegrityError
+from django.db.models import Q
 from django.http import Http404
 from django_filters import CharFilter, MultipleChoiceFilter
+from guardian.shortcuts import get_objects_for_user
 from drf_spectacular.utils import extend_schema
 from rest_framework import mixins
 from rest_framework.decorators import action
@@ -634,7 +636,7 @@ class ContainerPushRepositoryViewSet(TagOperationsMixin, ReadOnlyRepositoryViewS
     endpoint_name = "container-push"
     queryset = models.ContainerPushRepository.objects.all()
     serializer_class = serializers.ContainerPushRepositorySerializer
-    permission_classes = (AccessPolicyFromDB,)
+    permission_classes = (access_policy.NamespaceAccessPolicyFromDB,)
     queryset_filtering_required_permission = "container.view_containerpushrepository"
 
     DEFAULT_ACCESS_POLICY = {
@@ -648,29 +650,76 @@ class ContainerPushRepositoryViewSet(TagOperationsMixin, ReadOnlyRepositoryViewS
                 "action": ["retrieve"],
                 "principal": "authenticated",
                 "effect": "allow",
-                "condition": "has_model_or_obj_perms:container.view_containerpushrepository",
+                "condition": "has_namespace_or_obj_perms:container.view_containerpushrepository",
             },
             {
                 "action": ["tag", "untag"],
                 "principal": "authenticated",
                 "effect": "allow",
                 "condition": [
-                    "has_model_or_obj_perms:container.modify_content_containerpushrepository",
+                    "has_namespace_or_obj_perms:container.modify_content_containerpushrepository",
                 ],
             },
         ],
         "permissions_assignment": [
             {
-                "function": "add_for_object_creator",
-                "parameters": None,
+                "function": "add_perms_to_distribution_group",
+                "parameters": {
+                    "group_type": "owners",
+                    "add_user_to_group": True,
+                },
+                "permissions": [
+                    "container.view_containerpushrepository",
+                    "container.modify_content_containerpushrepository",
+                ],
+            },
+            {
+                "function": "add_perms_to_distribution_group",
+                "parameters": {
+                    "group_type": "collaborators",
+                    "add_user_to_group": False,
+                },
                 "permissions": [
                     "container.view_containerpushrepository",
                     "container.modify_content_containerpushrepository",
                 ],
             },
+            {
+                "function": "add_perms_to_distribution_group",
+                "parameters": {
+                    "group_type": "consumers",
+                    "add_user_to_group": False,
+                },
+                "permissions": [
+                    "container.view_containerpushrepository",
+                ],
+            },
         ],
     }
 
+    def get_queryset(self):
+        """
+        Returns a queryset by filtering by namespace permission to view distributions and
+        distribution level permissions.
+        """
+
+        qs = models.ContainerPushRepository.objects.all()
+        namespaces = get_objects_for_user(self.request.user, "container.view_containernamespace")
+        ns_repository_pks = models.ContainerDistribution.objects.filter(
+            namespace__in=namespaces
+        ).values_list("repository")
+        dist_repository_pks = get_objects_for_user(
+            self.request.user, "container.view_containerdistribution"
+        ).values_list("repository")
+        public_repository_pks = models.ContainerDistribution.objects.filter(
+            private=False
+        ).values_list("repository")
+        return qs.filter(
+            Q(pk__in=ns_repository_pks)
+            | Q(pk__in=dist_repository_pks)
+            | Q(pk__in=public_repository_pks)
+        )
+
 
 class ContainerPushRepositoryVersionViewSet(
     RepositoryVersionQuerySetMixin,
@@ -719,7 +768,7 @@ class ContainerDistributionViewSet(BaseDistributionViewSet):
     queryset = models.ContainerDistribution.objects.all()
     serializer_class = serializers.ContainerDistributionSerializer
     filterset_class = ContainerDistributionFilter
-    permission_classes = (access_policy.DistributionAccessPolicyFromDB,)
+    permission_classes = (access_policy.NamespaceAccessPolicyFromDB,)
     queryset_filtering_required_permission = "container.view_containerdistribution"
 
     DEFAULT_ACCESS_POLICY = {
@@ -738,17 +787,14 @@ class ContainerDistributionViewSet(BaseDistributionViewSet):
                 "action": ["create"],
                 "principal": "authenticated",
                 "effect": "allow",
-                "condition": [
-                    "has_model_perms:container.add_containerdistribution",
-                    "has_manage_namespace_dist_perms:container.manage_namespace_distributions",
-                ],
+                "condition": "has_model_perms:container.add_containerdistribution",
             },
             {
                 "action": ["retrieve"],
                 "principal": "authenticated",
                 "effect": "allow",
                 "condition": [
-                    "has_model_or_obj_perms:container.view_containerdistribution",
+                    "has_namespace_or_obj_perms:container.view_containerdistribution",
                 ],
             },
             {
@@ -764,7 +810,7 @@ class ContainerDistributionViewSet(BaseDistributionViewSet):
                 "principal": "authenticated",
                 "effect": "allow",
                 "condition": [
-                    "has_model_or_obj_perms:container.pull_containerdistribution",
+                    "has_namespace_or_obj_perms:container.pull_containerdistribution",
                 ],
             },
             {
@@ -780,7 +826,7 @@ class ContainerDistributionViewSet(BaseDistributionViewSet):
                 "principal": "authenticated",
                 "effect": "allow",
                 "condition": [
-                    "has_model_or_obj_perms:container.push_containerdistribution",
+                    "has_namespace_or_obj_perms:container.push_containerdistribution",
                     "obj_exists",
                 ],
             },
@@ -789,8 +835,8 @@ class ContainerDistributionViewSet(BaseDistributionViewSet):
                 "principal": "authenticated",
                 "effect": "allow",
                 "condition": [
-                    "has_model_perms:container.add_containerdistribution",
-                    "has_manage_namespace_dist_perms:container.manage_namespace_distributions",
+                    "has_namespace_or_obj_perms:container.add_containerdistribution",
+                    "has_namespace_or_obj_perms:container.push_containerdistribution",
                 ],
             },
             {
@@ -798,15 +844,17 @@ class ContainerDistributionViewSet(BaseDistributionViewSet):
                 "principal": "authenticated",
                 "effect": "allow",
                 "condition": [
-                    "has_model_or_obj_perms:container.delete_containerdistribution",
-                    "has_namespace_obj_perms:container.manage_namespace_distributions",
+                    "has_namespace_or_obj_perms:container.delete_containerdistribution",
                 ],
             },
         ],
         "permissions_assignment": [
             {
-                "function": "add_for_object_creator",
-                "parameters": None,
+                "function": "create_distribution_group",
+                "parameters": {
+                    "group_type": "owners",
+                    "add_user_to_group": True,
+                },
                 "permissions": [
                     "container.view_containerdistribution",
                     "container.pull_containerdistribution",
@@ -815,6 +863,58 @@ class ContainerDistributionViewSet(BaseDistributionViewSet):
                     "container.change_containerdistribution",
                 ],
             },
+            {
+                "function": "add_push_repository_perms_to_distribution_group",
+                "parameters": {
+                    "group_type": "owners",
+                },
+                "permissions": [
+                    "container.view_containerpushrepository",
+                    "container.modify_content_containerpushrepository",
+                ],
+            },
+            {
+                "function": "create_distribution_group",
+                "parameters": {
+                    "group_type": "collaborators",
+                    "add_user_to_group": False,
+                },
+                "permissions": [
+                    "container.view_containerdistribution",
+                    "container.pull_containerdistribution",
+                    "container.push_containerdistribution",
+                ],
+            },
+            {
+                "function": "add_push_repository_perms_to_distribution_group",
+                "parameters": {
+                    "group_type": "collaborators",
+                },
+                "permissions": [
+                    "container.view_containerpushrepository",
+                    "container.modify_content_containerpushrepository",
+                ],
+            },
+            {
+                "function": "create_distribution_group",
+                "parameters": {
+                    "group_type": "consumers",
+                    "add_user_to_group": False,
+                },
+                "permissions": [
+                    "container.view_containerdistribution",
+                    "container.pull_containerdistribution",
+                ],
+            },
+            {
+                "function": "add_push_repository_perms_to_distribution_group",
+                "parameters": {
+                    "group_type": "consumers",
+                },
+                "permissions": [
+                    "container.view_containerpushrepository",
+                ],
+            },
         ],
     }
 
@@ -896,15 +996,68 @@ class ContainerNamespaceViewSet(
                 "effect": "allow",
                 "condition": "has_model_or_obj_perms:container.delete_containernamespace",
             },
+            {
+                "action": ["create_distribution"],
+                "principal": "authenticated",
+                "effect": "allow",
+                "condition": "has_model_or_obj_perms:container.namespace_add_containerdistribution",
+            },
+            {
+                "action": ["view_distribution"],
+                "principal": "authenticated",
+                "effect": "allow",
+                "condition": "has_model_or_obj_perms:container.namespace_view_containerdistribution",  # noqa: E501
+            },
         ],
         "permissions_assignment": [
             {
-                "function": "add_for_object_creator",
-                "parameters": None,
+                "function": "create_namespace_group",
+                "parameters": {
+                    "group_type": "owners",
+                    "add_user_to_group": True,
+                },
                 "permissions": [
                     "container.view_containernamespace",
                     "container.delete_containernamespace",
-                    "container.manage_namespace_distributions",
+                    "container.namespace_add_containerdistribution",
+                    "container.namespace_delete_containerdistribution",
+                    "container.namespace_view_containerdistribution",
+                    "container.namespace_pull_containerdistribution",
+                    "container.namespace_push_containerdistribution",
+                    "container.namespace_change_containerdistribution",
+                    "container.namespace_view_containerpushrepository",
+                    "container.namespace_modify_content_containerpushrepository",
+                ],
+            },
+            {
+                "function": "create_namespace_group",
+                "parameters": {
+                    "group_type": "collaborators",
+                    "add_user_to_group": False,
+                },
+                "permissions": [
+                    "container.view_containernamespace",
+                    "container.namespace_add_containerdistribution",
+                    "container.namespace_delete_containerdistribution",
+                    "container.namespace_view_containerdistribution",
+                    "container.namespace_pull_containerdistribution",
+                    "container.namespace_push_containerdistribution",
+                    "container.namespace_change_containerdistribution",
+                    "container.namespace_view_containerpushrepository",
+                    "container.namespace_modify_content_containerpushrepository",
+                ],
+            },
+            {
+                "function": "create_namespace_group",
+                "parameters": {
+                    "group_type": "consumers",
+                    "add_user_to_group": False,
+                },
+                "permissions": [
+                    "container.view_containernamespace",
+                    "container.namespace_view_containerdistribution",
+                    "container.namespace_pull_containerdistribution",
+                    "container.namespace_view_containerpushrepository",
                 ],
             },
         ],
diff --git a/pulp_container/tests/functional/api/test_push_content.py b/pulp_container/tests/functional/api/test_push_content.py
index d7eaf8142..b00484280 100644
--- a/pulp_container/tests/functional/api/test_push_content.py
+++ b/pulp_container/tests/functional/api/test_push_content.py
@@ -8,7 +8,13 @@
 from pulp_smash.pulp3.bindings import monitor_task
 
 from pulp_container.tests.functional.constants import DOCKERHUB_PULP_FIXTURE_1
-from pulp_container.tests.functional.utils import del_user, gen_container_client, gen_user
+from pulp_container.tests.functional.utils import (
+    add_user_to_distribution_group,
+    add_user_to_namespace_group,
+    del_user,
+    gen_container_client,
+    gen_user,
+)
 
 from pulpcore.client.pulp_container import (
     PulpContainerNamespacesApi,
@@ -24,10 +30,6 @@ def setUpClass(cls):
         """
         Define APIs to use and pull images needed later in tests
         """
-        api_client = gen_container_client()
-        cls.pushrepository_api = RepositoriesContainerPushApi(api_client)
-        cls.namespace_api = PulpContainerNamespacesApi(api_client)
-
         cfg = config.get_config()
         cls.registry = cli.RegistryClient(cfg)
         cls.registry.raise_if_unsupported(unittest.SkipTest, "Tests require podman/docker")
@@ -36,36 +38,21 @@ def setUpClass(cls):
         admin_user, admin_password = cfg.pulp_auth
         cls.user_admin = {"username": admin_user, "password": admin_password}
         cls.user_creator = gen_user(
-            [
-                "container.add_containerdistribution",
-                "container.add_containernamespace",
-            ]
-        )
-        cls.user_dist_collaborator = gen_user(
-            [
-                "container.pull_containerdistribution",
-                "container.push_containerdistribution",
-                "container.view_containerpushrepository",
-            ]
-        )
-        cls.user_dist_consumer = gen_user(
-            [
-                "container.pull_containerdistribution",
-                "container.view_containerpushrepository",
-            ]
+            ["container.add_containernamespace", "container.add_containerdistribution"]
         )
-        cls.user_namespace_collaborator = gen_user(
-            [
-                "container.add_containerdistribution",
-                "container.pull_containerdistribution",
-                "container.push_containerdistribution",
-                "container.view_containerpushrepository",
-                "container.manage_namespace_distributions",
-            ]
-        )
-        cls.user_reader = gen_user(["container.view_containerpushrepository"])
+        cls.user_dist_collaborator = gen_user([])
+        cls.user_dist_consumer = gen_user([])
+        cls.user_namespace_collaborator = gen_user([])
+        cls.user_reader = gen_user([])
         cls.user_helpless = gen_user([])
 
+        # View push repositories, distributions, and namespaces using user_creator.
+        api_client = gen_container_client()
+        api_client.configuration.username = cls.user_admin["username"]
+        api_client.configuration.password = cls.user_admin["password"]
+        cls.pushrepository_api = RepositoriesContainerPushApi(api_client)
+        cls.namespace_api = PulpContainerNamespacesApi(api_client)
+
         cls._pull(f"{DOCKERHUB_PULP_FIXTURE_1}:manifest_a")
         cls._pull(f"{DOCKERHUB_PULP_FIXTURE_1}:manifest_b")
         cls._pull(f"{DOCKERHUB_PULP_FIXTURE_1}:manifest_c")
@@ -156,6 +143,29 @@ def test_push_with_dist_perms(self):
         image_path = f"{DOCKERHUB_PULP_FIXTURE_1}:manifest_a"
         self._push(image_path, local_url, self.user_creator)
 
+        distributions = self.user_creator["distribution_api"].list(name="test/perms")
+        add_user_to_distribution_group(
+            self.user_dist_collaborator,
+            distributions.results[0],
+            "collaborators",
+            self.user_creator,
+        )
+
+        distributions = self.user_creator["distribution_api"].list(name="test/perms")
+        add_user_to_distribution_group(
+            self.user_dist_consumer,
+            distributions.results[0],
+            "consumers",
+            self.user_creator,
+        )
+
+        add_user_to_namespace_group(
+            self.user_namespace_collaborator,
+            "test",
+            "collaborators",
+            self.user_creator,
+        )
+
         self.assertEqual(self.pushrepository_api.list(name=repo_name).count, 1)
         self.assertEqual(self.user_creator["pushrepository_api"].list(name=repo_name).count, 1)
         self.assertEqual(
@@ -167,8 +177,8 @@ def test_push_with_dist_perms(self):
         self.assertEqual(
             self.user_namespace_collaborator["pushrepository_api"].list(name=repo_name).count, 1
         )
+
         self.assertEqual(self.user_reader["pushrepository_api"].list(name=repo_name).count, 1)
-        self.assertEqual(self.user_helpless["pushrepository_api"].list(name=repo_name).count, 0)
 
         # cleanup, namespace removal also removes related distributions
         namespace = self.namespace_api.list(name="test").results[0]
@@ -218,6 +228,15 @@ def test_push_to_existing_namespace(self):
         image_path = f"{DOCKERHUB_PULP_FIXTURE_1}:manifest_a"
         self._push(image_path, local_url, self.user_creator)
 
+        # Add user_dist_collaborator to the collaborator group
+        distributions = self.user_creator["distribution_api"].list(name="team/owner")
+        add_user_to_distribution_group(
+            self.user_dist_collaborator,
+            distributions.results[0],
+            "collaborators",
+            self.user_creator,
+        )
+
         collab_repo_name = "team/owner"
         local_url = "/".join([self.registry_name, f"{collab_repo_name}:2.0"])
         image_path = f"{DOCKERHUB_PULP_FIXTURE_1}:manifest_b"
@@ -229,6 +248,10 @@ def test_push_to_existing_namespace(self):
         with self.assertRaises(exceptions.CalledProcessError):
             self._push(image_path, local_url, self.user_dist_collaborator)
 
+        add_user_to_namespace_group(
+            self.user_namespace_collaborator, "team", "collaborators", self.user_creator
+        )
+
         collab_repo_name = "team/collab"
         local_url = "/".join([self.registry_name, f"{collab_repo_name}:2.0"])
         image_path = f"{DOCKERHUB_PULP_FIXTURE_1}:manifest_c"
@@ -264,6 +287,11 @@ def test_private_repository(self):
         self._push(image_path, local_url, self.user_creator)
 
         self._pull(local_url, self.user_creator)
+
+        add_user_to_distribution_group(
+            self.user_dist_consumer, distribution, "consumers", self.user_creator
+        )
+
         self._pull(local_url, self.user_dist_consumer)
         with self.assertRaises(exceptions.CalledProcessError):
             self._pull(local_url, self.user_reader)
diff --git a/pulp_container/tests/functional/utils.py b/pulp_container/tests/functional/utils.py
index 9a947d93c..5b846fd61 100644
--- a/pulp_container/tests/functional/utils.py
+++ b/pulp_container/tests/functional/utils.py
@@ -25,6 +25,8 @@
 from pulpcore.client.pulpcore import (
     ApiClient as CoreApiClient,
     ArtifactsApi,
+    GroupsApi,
+    GroupsUsersApi,
     TasksApi,
 )
 from pulpcore.client.pulp_container import (
@@ -80,6 +82,9 @@ def gen_user(permissions):
     api_config = cfg.get_bindings_config()
     api_config.username = user["username"]
     api_config.password = user["password"]
+    user["core_api_client"] = CoreApiClient(api_config)
+    user["groups_api"] = GroupsApi(user["core_api_client"])
+    user["group_users_api"] = GroupsUsersApi(user["core_api_client"])
     user["api_client"] = ContainerApiClient(api_config)
     user["distribution_api"] = DistributionsContainerApi(user["api_client"])
     user["remote_api"] = RemotesContainerApi(user["api_client"])
@@ -97,6 +102,30 @@ def del_user(user):
     )
 
 
+def add_user_to_distribution_group(user, distribution, group_type, as_user):
+    """Add the user to either owner, collaborator, or consumer group of a distribution."""
+    distribution_pk = distribution.pulp_href.split("/")[-2]
+    collaborator_group = (
+        as_user["groups_api"]
+        .list(name="container.distribution.{}.{}".format(group_type, distribution_pk))
+        .results[0]
+    )
+    as_user["group_users_api"].create(collaborator_group.pulp_href, {"username": user["username"]})
+
+
+def add_user_to_namespace_group(user, namespace_name, group_type, as_user):
+    """Add the user to either owner, collaborator, or consumer group of a namespace."""
+    namespace_collaborator_group = (
+        as_user["groups_api"]
+        .list(name="container.namespace.{}.{}".format(group_type, namespace_name))
+        .results[0]
+    )
+    as_user["group_users_api"].create(
+        namespace_collaborator_group.pulp_href,
+        {"username": user["username"]},
+    )
+
+
 def gen_container_client():
     """Return an OBJECT for container client."""
     return ContainerApiClient(configuration)
diff --git a/pulp_container/tests/unit/test_serializers.py b/pulp_container/tests/unit/test_serializers.py
index 6be261b7c..b63426ec9 100644
--- a/pulp_container/tests/unit/test_serializers.py
+++ b/pulp_container/tests/unit/test_serializers.py
@@ -1,5 +1,8 @@
+from django.contrib.auth import get_user_model
 from django.test import TestCase
 
+from django_currentuser.middleware import _set_current_user
+
 from pulp_container.app.serializers import ContainerDistributionSerializer, TagOperationSerializer
 from pulp_container.app.models import ContainerPushRepository, ContainerRepository
 
@@ -21,6 +24,14 @@ def setUp(self):
         self.push_repository_href = "/pulp/api/v3/repositories/container/container-push/{}/".format(
             self.push_repository.pk
         )
+        self.user = get_user_model().objects.create(username="user1", is_staff=False)
+        _set_current_user(self.user)
+
+    def tearDown(self):
+        """Delete the user."""
+        super().tearDown()
+        self.user.delete()
+        _set_current_user(None)
 
     def test_valid_mirror_data(self):
         """Test that the ContainerDistributionSerializer accepts valid data."""