Lock out plugins when in GPOS STIG mode

Certain TrueNAS features are incompatibile with requirements for
the GPOS STIG. This commit makes changes to our RBAC framework
to decrease effective privileges granted to all credentials to
prevent configuring non-compliant services and to give UI hints
about which features to disable.

As a result of these changes, the allowlist column is dropped
from the privilege table. Internally our credentials still
generate an allowlist in order to perform RBAC checks.

The allowlist changes have a functional impact on privilege
framework regarding REST API access. After these changes,
only credentials with FULL_ADMIN privilege will be allowed
REST access and only when server does not have STIG rules
applied.

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")
 
@@ -192,7 +193,7 @@ async def _validate(self, schema_name, data, id_=None):
 
         for i, role in enumerate(data["roles"]):
             if role not in self.middleware.role_manager.roles:
-                verrors.add(f"{schema_name}.roles.{i}", "Invalid role")
+                verrors.add(f"{schema_name}.roles.{i}", f"Invalid role: choices are {self.middleware.role_manager.roles.keys()}")
 
         verrors.check()
 
@@ -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,19 @@
+from middlewared.alert.base import (
+    AlertCategory, AlertClass, AlertLevel, OneShotAlertClass, 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 +38,34 @@ 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