NAS-133042 / 25.04 / Lock out plugins when in GPOS STIG mode

src/middlewared/middlewared/alembic/versions/25.04/2024-12-11_20-52_remove-privilege-allowlist.py

@@ -0,0 +1,32 @@
+""" Remove privilege allowlist
+
+Revision ID: aea6bced4328
+Revises: b44c092bfa30
+Create Date: 2024-12-11 20:52:26.972597+00:00
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = 'aea6bced4328'
+down_revision = 'b44c092bfa30'
+branch_labels = None
+depends_on = None
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.execute("UPDATE account_privilege SET roles='[\"FULL_ADMIN\"]' WHERE allowlist='[{\"method\": \"*\", \"resource\": \"*\"}]'")
+    with op.batch_alter_table('account_privilege', schema=None) as batch_op:
+        batch_op.drop_column('allowlist')
+
+    # ### end Alembic commands ###
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    with op.batch_alter_table('account_privilege', schema=None) as batch_op:
+        batch_op.add_column(sa.Column('allowlist', sa.TEXT(), nullable=False))
+
+    # ### end Alembic commands ###

src/middlewared/middlewared/api/base/decorator.py

@@ -92,7 +92,7 @@ def wrapped(*args):
         wrapped.audit_callback = audit_callback
         wrapped.audit_extended = audit_extended
         wrapped.rate_limit = rate_limit
-        wrapped.roles = roles or []
+        wrapped.roles = roles or ['FULL_ADMIN']
         wrapped._private = private
         wrapped._cli_private = cli_private
 

src/middlewared/middlewared/api/v25_04_0/privilege.py

@@ -1,8 +1,8 @@
 from middlewared.api.base import BaseModel, Excluded, excluded_field, ForUpdateMetaclass, NonEmptyString, SID
-from .api_key import AllowListItem
+from middlewared.utils.security import STIGType
 from .group import GroupEntry
 
-__all__ = ["PrivilegeEntry",
+__all__ = ["PrivilegeEntry", "PrivilegeRoleEntry",
            "PrivilegeCreateArgs", "PrivilegeCreateResult",
            "PrivilegeUpdateArgs", "PrivilegeUpdateResult",
            "PrivilegeDeleteArgs", "PrivilegeDeleteResult"]
@@ -14,7 +14,6 @@ class PrivilegeEntry(BaseModel):
     name: NonEmptyString
     local_groups: list[GroupEntry]
     ds_groups: list[GroupEntry]
-    allowlist: list[AllowListItem] = []
     roles: list[str] = []
     web_shell: bool
 
@@ -53,3 +52,11 @@ class PrivilegeDeleteArgs(BaseModel):
 
 class PrivilegeDeleteResult(BaseModel):
     result: bool
+
+
+class PrivilegeRoleEntry(BaseModel):
+    name: NonEmptyString
+    title: NonEmptyString
+    includes: list[NonEmptyString]
+    builtin: bool
+    stig: STIGType | None

src/middlewared/middlewared/plugins/account.py

@@ -1291,6 +1291,13 @@ async def common_validation(self, verrors, data, schema, group_ids, old=None):
                 f'{schema}.password_disabled', 'Password authentication may not be disabled for SMB users.'
             )
 
+        if combined['smb'] and (await self.middleware.call('system.security.config'))['enable_gpos_stig']:
+            verrors.add(
+                f'{schema}.smb',
+                'SMB authentication for local user accounts is not permitted when General Purpose OS '
+                'STIG compatibility is enabled.'
+            )
+
         password = data.get('password')
         if not old and not password and not data.get('password_disabled'):
             verrors.add(f'{schema}.password', 'Password is required')
@@ -1593,10 +1600,6 @@ async def group_extend(self, group, ctx):
         group['users'] = list({u['id'] for u in group['users']} | ctx['primary_memberships'][group['id']])
 
         privilege_mappings = privileges_group_mapping(ctx['privileges'], [group['gid']], 'local_groups')
-        if privilege_mappings['allowlist']:
-            privilege_mappings['roles'].append('HAS_ALLOW_LIST')
-            if {'method': '*', 'resource': '*'} in privilege_mappings['allowlist']:
-                privilege_mappings['roles'].append('FULL_ADMIN')
 
         match group['group']:
             case 'builtin_administrators':

src/middlewared/middlewared/plugins/account_/privilege.py

@@ -12,6 +12,7 @@
     privilege_has_webui_access,
     privileges_group_mapping
 )
+from middlewared.utils.security import system_security_config_to_stig_type
 import middlewared.sqlalchemy as sa
 
 
@@ -29,7 +30,6 @@ class PrivilegeModel(sa.Model):
     name = sa.Column(sa.String(200))
     local_groups = sa.Column(sa.JSON(list))
     ds_groups = sa.Column(sa.JSON(list))
-    allowlist = sa.Column(sa.JSON(list))
     roles = sa.Column(sa.JSON(list))
     web_shell = sa.Column(sa.Boolean())
 
@@ -45,6 +45,7 @@ class Config:
         datastore_extend_context = "privilege.item_extend_context"
         cli_namespace = "auth.privilege"
         entry = PrivilegeEntry
+        role_prefix = 'PRIVILEGE'
 
     @private
     async def item_extend_context(self, rows, extra):
@@ -70,7 +71,7 @@ async def do_create(self, data):
 
         `ds_groups` is list of Directory Service group GIDs that will gain this privilege.
 
-        `allowlist` is a list of API endpoints allowed for this privilege.
+        `roles` is a list of roles to be assigned to the privilege
 
         `web_shell` controls whether users with this privilege are allowed to log in to the Web UI.
         """
@@ -105,7 +106,7 @@ async def do_update(self, audit_callback, id_, data):
         verrors = ValidationErrors()
 
         if new["builtin_name"]:
-            for k in ["name", "allowlist", "roles"]:
+            for k in ["name", "roles"]:
                 if new[k] != old[k]:
                     verrors.add(f"privilege_update.{k}", "This field is read-only for built-in privileges")
 
@@ -354,6 +355,9 @@ async def privileges_for_groups(self, groups_key, group_ids):
 
     @private
     async def compose_privilege(self, privileges):
+        security_config = await self.middleware.call('system.security.config')
+        enabled_stig = system_security_config_to_stig_type(security_config)
+
         compose = {
             'roles': set(),
             'allowlist': [],
@@ -362,20 +366,16 @@ async def compose_privilege(self, privileges):
         }
         for privilege in privileges:
             for role in privilege['roles']:
-                compose['roles'] |= self.middleware.role_manager.roles_for_role(role)
-
-                compose['allowlist'].extend(self.middleware.role_manager.allowlist_for_role(role))
+                compose['roles'] |= self.middleware.role_manager.roles_for_role(role, enabled_stig)
 
-            for item in privilege['allowlist']:
-                if item == {'method': '*', 'resource': '*'} and 'FULL_ADMIN' not in compose['roles']:
-                    compose['roles'] |= self.middleware.role_manager.roles_for_role('FULL_ADMIN')
-                    compose['webui_access'] = True
-
-                compose['allowlist'].append(item)
+                compose['allowlist'].extend(self.middleware.role_manager.allowlist_for_role(role, enabled_stig))
 
             compose['web_shell'] |= privilege['web_shell']
             compose['webui_access'] |= privilege_has_webui_access(privilege)
 
+        if enabled_stig:
+            compose['web_shell'] = False
+
         return compose
 
     @private

src/middlewared/middlewared/plugins/account_/privilege_roles.py

@@ -1,6 +1,7 @@
+from middlewared.api.current import PrivilegeRoleEntry
+
 from middlewared.role import ROLES
-from middlewared.service import Service, filterable, filterable_returns, filter_list, no_authz_required
-from middlewared.schema import Bool, Dict, List, Str
+from middlewared.service import Service, filterable_api_method, filter_list
 
 
 class PrivilegeService(Service):
@@ -9,15 +10,7 @@ class Config:
         namespace = "privilege"
         cli_namespace = "auth.privilege"
 
-    @no_authz_required
-    @filterable
-    @filterable_returns(Dict(
-        "role",
-        Str("name"),
-        Str("title"),
-        List("includes", items=[Str("name")]),
-        Bool("builtin")
-    ))
+    @filterable_api_method(item=PrivilegeRoleEntry, authorization_required=False)
     async def roles(self, filters, options):
         """
         Get all available roles.
@@ -39,6 +32,7 @@ async def roles(self, filters, options):
                 "title": name,
                 "includes": role.includes,
                 "builtin": role.builtin,
+                "stig": role.stig,
             }
             for name, role in ROLES.items()
         ]

src/middlewared/middlewared/plugins/directoryservices_/activedirectory_join_mixin.py

@@ -251,7 +251,7 @@ def _ad_grant_privileges(self) -> None:
             self.middleware.call_sync('privilege.create', {
                 'name': dom.dns_name.upper(),
                 'ds_groups': [f'{dom.sid}-512'],
-                'allowlist': [{'method': '*', 'resource': '*'}],
+                'roles': ['FULL_ADMIN'],
                 'web_shell': True
             })
         except Exception:

src/middlewared/middlewared/plugins/security/update.py

@@ -24,7 +24,7 @@ class Config:
         cli_namespace = 'system.security'
         datastore = 'system.security'
         namespace = 'system.security'
-        role = 'SYSTEM_SECURITY'
+        role_prefix = 'SYSTEM_SECURITY'
         entry = SystemSecurityEntry
 
     @private
@@ -188,6 +188,8 @@ async def do_update(self, job, data):
                 )
                 await self.configure_security_on_ha(is_ha, job, RebootReason.GPOSSTIG)
 
+            await self.configure_stig(new)
+
         return await self.config()
 
 

src/middlewared/middlewared/plugins/test/mock.py

@@ -1,6 +1,17 @@
+from middlewared.alert.base import  AlertCategory, AlertClass, AlertLevel, SimpleOneShotAlertClass
+from middlewared.role import Role
 from middlewared.service import CallError, Service
 
 
+class SystemTestingAlertClass(AlertClass, SimpleOneShotAlertClass):
+    category = AlertCategory.SYSTEM
+    level = AlertLevel.CRITICAL
+    title = "System mocking endpoints used"
+    text = "System mocking endpoits used on server."
+
+    deleted_automatically = False
+
+
 class TestService(Service):
     class Config:
         private = True
@@ -25,12 +36,31 @@ def method(*args, **kwargs):
         else:
             raise CallError("Invalid mock declaration")
 
+        await self.set_mock_role()
         self.middleware.set_mock(name, args, method)
 
     async def remove_mock(self, name, args):
         self.middleware.remove_mock(name, args)
 
+    async def set_mock_role(self):
+        """
+        adds a MOCK role to role_manager and grants access to test.test1 and test.test2
+
+        This allows testing RBAC against mocked endpoint
+        """
+        if 'MOCK' in self.middleware.role_manager.roles:
+            return
+
+        # There are no STIG requirements specified for MOCK role here because
+        # we need to be able to mock methods in CI testing while under STIG restrictions
+        self.middleware.role_manager.roles['MOCK'] = Role()
+        self.middleware.role_manager.register_method(method_name='test.test1', roles=['MOCK'])
+        self.middleware.role_manager.register_method(method_name='test.test2', roles=['MOCK'])
+
+        await self.middleware.call('alert.oneshot_create', 'SystemTesting')
+
     # Dummy methods to mock for internal infrastructure testing (i.e. jobs manager)
+    # When these are mocked over they will be available to users with the "MOCK" role.
 
     async def test1(self):
         pass