diff --git a/trustpoint/agents/api_views.py b/trustpoint/agents/api_views.py index 208e7c215..ac289aa7e 100644 --- a/trustpoint/agents/api_views.py +++ b/trustpoint/agents/api_views.py @@ -21,6 +21,7 @@ from rest_framework.views import APIView from agents.models import AgentAssignedProfile, TrustpointAgent +from agents.security import AgentSecurity from devices.models import DeviceModel from trustpoint.logger import LoggerMixin @@ -260,6 +261,12 @@ class AgentJobsView(LoggerMixin, APIView): ) def get(self, request: Request) -> Response: """Return pending renewal jobs for the authenticated agent.""" + if not AgentSecurity.is_agent_protocol_permitted(): + return Response( + {'detail': 'Agent functionality is disabled by the current security configuration.'}, + status=status.HTTP_403_FORBIDDEN, + ) + agent: TrustpointAgent = request.user # type: ignore[assignment] TrustpointAgent.objects.filter(pk=agent.pk).update(last_seen_at=timezone.now()) @@ -393,6 +400,12 @@ class AgentJobResultView(LoggerMixin, APIView): ) def post(self, request: Request) -> Response: """Process a job result posted by the authenticated agent.""" + if not AgentSecurity.is_agent_protocol_permitted(): + return Response( + {'detail': 'Agent functionality is disabled by the current security configuration.'}, + status=status.HTTP_403_FORBIDDEN, + ) + agent: TrustpointAgent = request.user # type: ignore[assignment] TrustpointAgent.objects.filter(pk=agent.pk).update(last_seen_at=timezone.now()) diff --git a/trustpoint/agents/security.py b/trustpoint/agents/security.py new file mode 100644 index 000000000..562b612b2 --- /dev/null +++ b/trustpoint/agents/security.py @@ -0,0 +1,58 @@ +# Copyright (c) 2026 The Trustpoint Project Authors +# SPDX-License-Identifier: MIT + +"""Security utilities for the agents application.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Any + +from django.contrib import messages +from django.http import HttpResponseRedirect +from django.urls import reverse_lazy + +from onboarding.enums import OnboardingProtocol +from trustpoint.logger import LoggerMixin + +if TYPE_CHECKING: + from django.http import HttpRequest, HttpResponse + + +class AgentSecurity(LoggerMixin): + """Helper class for checking agent-related security policies.""" + + @staticmethod + def is_agent_protocol_permitted() -> bool: + """Check if the AGENT onboarding protocol is permitted by the active security configuration. + + Returns: + True if AGENT protocol is permitted, False otherwise. + """ + from management.models import SecurityConfig # noqa: PLC0415 + + try: + cfg: SecurityConfig = SecurityConfig.objects.get() + except SecurityConfig.DoesNotExist: + return True + except SecurityConfig.MultipleObjectsReturned: + cfg = SecurityConfig.objects.first() # type: ignore[assignment] + + permitted: list[int] = cfg.permitted_onboarding_protocols or [] + return OnboardingProtocol.AGENT.value in permitted + + +class AgentSecurityMixin(LoggerMixin): + """Mixin for views that require the AGENT protocol to be permitted.""" + + def dispatch(self, request: HttpRequest, *args: Any, **kwargs: Any) -> HttpResponse: + """Check if AGENT protocol is permitted before dispatching the view.""" + if not AgentSecurity.is_agent_protocol_permitted(): + self.logger.warning( + 'Access denied to agent functionality: AGENT protocol not permitted by security configuration.' + ) + messages.error( + request, + 'Agent functionality is disabled by the current security configuration.', + ) + return HttpResponseRedirect(reverse_lazy('devices:list')) + return super().dispatch(request, *args, **kwargs) # type: ignore[misc,no-any-return] diff --git a/trustpoint/agents/web_views.py b/trustpoint/agents/web_views.py index 9023e6b8e..b9a8dd3a9 100644 --- a/trustpoint/agents/web_views.py +++ b/trustpoint/agents/web_views.py @@ -18,6 +18,7 @@ from django.views.generic import FormView, ListView, UpdateView from agents.models import AgentAssignedProfile, AgentProfileDefinition, TrustpointAgent +from agents.security import AgentSecurityMixin from trustpoint.logger import LoggerMixin from trustpoint.page_context import DEVICES_PAGE_AGENTS_SUBCATEGORY, DEVICES_PAGE_CATEGORY, PageContextMixin from trustpoint.views.base import BulkDeleteView @@ -27,7 +28,9 @@ -class AgentProfileDefinitionTableView(PageContextMixin, LoggerMixin, ListView[AgentProfileDefinition]): +class AgentProfileDefinitionTableView( + AgentSecurityMixin, PageContextMixin, LoggerMixin, ListView[AgentProfileDefinition] +): """View to list all Agent Profile Definitions.""" http_method_names = ('get',) @@ -43,7 +46,9 @@ def get_queryset(self) -> QuerySet[AgentProfileDefinition]: return AgentProfileDefinition.objects.all().order_by('name') -class AgentProfileDefinitionConfigView(PageContextMixin, LoggerMixin, UpdateView[AgentProfileDefinition, Any]): +class AgentProfileDefinitionConfigView( + AgentSecurityMixin, PageContextMixin, LoggerMixin, UpdateView[AgentProfileDefinition, Any] +): """View to display and edit an Agent Profile Definition.""" http_method_names = ('get', 'post') @@ -199,7 +204,7 @@ def form_valid(self, form: Any) -> Any: return super().form_valid(form) -class AgentProfileDefinitionBulkDeleteConfirmView(PageContextMixin, BulkDeleteView): +class AgentProfileDefinitionBulkDeleteConfirmView(AgentSecurityMixin, PageContextMixin, BulkDeleteView): """View to confirm the deletion of multiple workflow definitions.""" model = AgentProfileDefinition @@ -293,7 +298,9 @@ class Meta: } -class AgentAssignedProfileTableView(PageContextMixin, LoggerMixin, ListView[AgentAssignedProfile]): +class AgentAssignedProfileTableView( + AgentSecurityMixin, PageContextMixin, LoggerMixin, ListView[AgentAssignedProfile] +): """List all workflow profiles assigned to a specific 1-to-1 agent.""" http_method_names: ClassVar[list[str]] = ['get'] # type: ignore[misc] @@ -319,7 +326,9 @@ def get_context_data(self, **kwargs: Any) -> dict[str, Any]: return context -class AgentAssignedProfileCreateView(PageContextMixin, LoggerMixin, FormView[AgentAssignedProfileForm]): +class AgentAssignedProfileCreateView( + AgentSecurityMixin, PageContextMixin, LoggerMixin, FormView[AgentAssignedProfileForm] +): """Assign a new workflow profile to a 1-to-1 agent.""" http_method_names: ClassVar[list[str]] = ['get', 'post'] # type: ignore[misc] @@ -362,7 +371,7 @@ def form_valid(self, form: AgentAssignedProfileForm) -> HttpResponse: class AgentAssignedProfileEditView( - PageContextMixin, LoggerMixin, UpdateView[AgentAssignedProfile, AgentAssignedProfileEditForm] + AgentSecurityMixin, PageContextMixin, LoggerMixin, UpdateView[AgentAssignedProfile, AgentAssignedProfileEditForm] ): """Edit an existing AgentAssignedProfile (renewal_threshold_days, subject, subject_alt_name).""" @@ -400,7 +409,7 @@ def form_valid(self, form: AgentAssignedProfileEditForm) -> HttpResponse: return HttpResponseRedirect(self.get_success_url()) -class AgentAssignedProfileDeleteView(PageContextMixin, BulkDeleteView): +class AgentAssignedProfileDeleteView(AgentSecurityMixin, PageContextMixin, BulkDeleteView): """Confirm and execute bulk deletion of AgentAssignedProfile records.""" model = AgentAssignedProfile @@ -445,7 +454,7 @@ def form_valid(self, form: Any) -> HttpResponse: return response -class AgentAssignedProfileForceUpdateView(PageContextMixin, LoggerMixin, View): +class AgentAssignedProfileForceUpdateView(AgentSecurityMixin, PageContextMixin, LoggerMixin, View): """Force an immediate certificate update by setting next_certificate_update_scheduled to now.""" http_method_names: ClassVar[list[str]] = ['post'] # type: ignore[misc] diff --git a/trustpoint/devices/forms.py b/trustpoint/devices/forms.py index 62648aef1..5cea56742 100644 --- a/trustpoint/devices/forms.py +++ b/trustpoint/devices/forms.py @@ -64,6 +64,33 @@ class SwitchCheckboxSelectMultiple(forms.CheckboxSelectMultiple): ] +def _get_permitted_onboarding_protocols() -> list[tuple[int, Any]]: + """Return the list of onboarding protocols permitted by the current security configuration. + + Returns: + List of tuples (protocol_value, protocol_label) for permitted protocols. + """ + from management.models import SecurityConfig # noqa: PLC0415 + + try: + cfg: SecurityConfig = SecurityConfig.objects.get() + permitted: list[int] = cfg.permitted_onboarding_protocols or [] + except SecurityConfig.DoesNotExist: + return ONBOARDING_PROTOCOLS_ALLOWED_FOR_FORMS + except SecurityConfig.MultipleObjectsReturned: + cfg = SecurityConfig.objects.first() # type: ignore[assignment] + permitted = cfg.permitted_onboarding_protocols or [] if cfg else [] + + if not permitted: + return [] + + return [ + (proto_value, proto_label) + for proto_value, proto_label in ONBOARDING_PROTOCOLS_ALLOWED_FOR_FORMS + if proto_value in permitted + ] + + def _get_secret(number_of_symbols: int = 16) -> str: """Generates a secret with the number of symbols provided. @@ -492,6 +519,24 @@ def __init__(self, *args: Any, **kwargs: Any) -> None: """Initializes the CreateDeviceForm.""" super().__init__(*args, **kwargs) + permitted_protocols = _get_permitted_onboarding_protocols() + onboarding_protocol_field = self.fields['onboarding_protocol'] + if isinstance(onboarding_protocol_field, forms.ChoiceField): + onboarding_protocol_field.choices = permitted_protocols + + all_protocol_values = {proto[0] for proto in permitted_protocols} + disabled_options = [ + proto for proto in [ + OnboardingProtocol.MANUAL, + OnboardingProtocol.AOKI, + OnboardingProtocol.BRSKI, + OnboardingProtocol.OPC_GDS_PUSH, + ] + if proto.value in all_protocol_values + ] + if isinstance(onboarding_protocol_field, forms.ChoiceField): + onboarding_protocol_field.widget = DisableOptionsSelect(disabled_options=disabled_options) + self.helper = FormHelper() self.helper.form_tag = False @@ -611,16 +656,18 @@ class AgentOnboardingCreateForm(OnboardingCreateForm): onboarding_protocol = forms.ChoiceField( choices=[(OnboardingProtocol.REST_USERNAME_PASSWORD.value, OnboardingProtocol.REST_USERNAME_PASSWORD.label)], - initial=OnboardingProtocol.REST_USERNAME_PASSWORD, + initial=OnboardingProtocol.REST_USERNAME_PASSWORD.value, label=_('Onboarding Protocol'), widget=forms.HiddenInput(), + required=True, ) # REST is the only allowed PKI protocol for agents — submitted as hidden input. onboarding_pki_protocols = forms.MultipleChoiceField( - choices=[(OnboardingPkiProtocol.REST, OnboardingPkiProtocol.REST.label)], - initial=[OnboardingPkiProtocol.REST], + choices=[(OnboardingPkiProtocol.REST.value, OnboardingPkiProtocol.REST.label)], + initial=[OnboardingPkiProtocol.REST.value], widget=forms.MultipleHiddenInput(), + required=True, ) # Agent-specific field for OS path where certificates will be stored @@ -637,11 +684,24 @@ class AgentOnboardingCreateForm(OnboardingCreateForm): def __init__(self, *args: Any, **kwargs: Any) -> None: """Initializes the AgentOnboardingCreateForm with fixed EST-only fields.""" + if 'initial' not in kwargs: + kwargs['initial'] = {} + kwargs['initial']['onboarding_protocol'] = OnboardingProtocol.REST_USERNAME_PASSWORD.value + kwargs['initial']['onboarding_pki_protocols'] = [OnboardingPkiProtocol.REST.value] + super().__init__(*args, **kwargs) - # Ensure hidden fields carry the correct pre-selected values. - self.initial['onboarding_protocol'] = str(OnboardingProtocol.REST_USERNAME_PASSWORD.value) - self.initial['onboarding_pki_protocols'] = [str(OnboardingPkiProtocol.REST.value)] + onboarding_protocol_field = self.fields['onboarding_protocol'] + if isinstance(onboarding_protocol_field, forms.ChoiceField): + onboarding_protocol_field.choices = [ + (OnboardingProtocol.REST_USERNAME_PASSWORD.value, OnboardingProtocol.REST_USERNAME_PASSWORD.label) + ] + + onboarding_pki_protocols_field = self.fields['onboarding_pki_protocols'] + if isinstance(onboarding_pki_protocols_field, forms.MultipleChoiceField): + onboarding_pki_protocols_field.choices = [ + (OnboardingPkiProtocol.REST.value, OnboardingPkiProtocol.REST.label) + ] self.helper = FormHelper() self.helper.form_tag = False diff --git a/trustpoint/management/models/security.py b/trustpoint/management/models/security.py index fa6d58ea4..6c58f02ca 100644 --- a/trustpoint/management/models/security.py +++ b/trustpoint/management/models/security.py @@ -214,9 +214,9 @@ class HashAlgorithmChoices(models.TextChoices): # ------------------------------------------------------------------ #: All OnboardingProtocol values - _ALL_ONBOARDING_PROTOCOLS: ClassVar[list[int]] = [0, 1, 2, 3, 4, 5, 6, 7, 8] + _ALL_ONBOARDING_PROTOCOLS: ClassVar[list[int]] = [0, 1, 2, 3, 4, 5, 6, 7, 8, 9] #: All OnboardingProtocol values except MANUAL (0) - _ONBOARDING_PROTOCOLS_NO_MANUAL: ClassVar[list[int]] = [1, 2, 3, 4, 5, 6, 7, 8] + _ONBOARDING_PROTOCOLS_NO_MANUAL: ClassVar[list[int]] = [1, 2, 3, 4, 5, 6, 7, 8, 9] _MODE_DEFAULTS: ClassVar[dict[str, _SecurityModeDefaults]] = { # ----------------------------------------------------------------