Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 13 additions & 0 deletions trustpoint/agents/api_views.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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,
)
Comment on lines +264 to +268

agent: TrustpointAgent = request.user # type: ignore[assignment]

TrustpointAgent.objects.filter(pk=agent.pk).update(last_seen_at=timezone.now())
Expand Down Expand Up @@ -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())
Expand Down
58 changes: 58 additions & 0 deletions trustpoint/agents/security.py
Original file line number Diff line number Diff line change
@@ -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]
35 changes: 24 additions & 11 deletions trustpoint/agents/web_views.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,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
Expand All @@ -29,7 +30,9 @@
from devices.models import DeviceModel


class AgentProfileDefinitionTableView(PageContextMixin, LoggerMixin, ListView[AgentProfileDefinition]):
class AgentProfileDefinitionTableView(
AgentSecurityMixin, PageContextMixin, LoggerMixin, ListView[AgentProfileDefinition]
):
Comment on lines +33 to +35
"""View to list all Agent Profile Definitions."""

http_method_names = ('get',)
Expand All @@ -45,7 +48,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')
Expand Down Expand Up @@ -201,7 +206,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
Expand Down Expand Up @@ -238,7 +243,9 @@ def form_valid(self, form: Any) -> HttpResponse:
return response


class AgentManagedDeviceTableView(PageContextMixin, LoggerMixin, ListView['DeviceModel']):
class AgentManagedDeviceTableView(
AgentSecurityMixin, PageContextMixin, LoggerMixin, ListView['DeviceModel']
):
"""List all AGENT_MANAGED_DEVICE devices in the same domain as a 1-to-n agent's device."""

http_method_names: ClassVar[list[str]] = ['get'] # type: ignore[misc]
Expand Down Expand Up @@ -296,7 +303,9 @@ class ManagedDeviceCreateForm(forms.Form):
)


class AgentManagedDeviceCreateView(PageContextMixin, LoggerMixin, FormView[ManagedDeviceCreateForm]):
class AgentManagedDeviceCreateView(
AgentSecurityMixin, PageContextMixin, LoggerMixin, FormView[ManagedDeviceCreateForm]
):
"""Create a new AGENT_MANAGED_DEVICE record under a 1-to-n agent."""

http_method_names: ClassVar[list[str]] = ['get', 'post'] # type: ignore[misc]
Expand Down Expand Up @@ -368,7 +377,7 @@ def form_valid(self, form: ManagedDeviceCreateForm) -> HttpResponse:
return HttpResponseRedirect(self.get_success_url())


class AgentManagedDeviceDeleteView(PageContextMixin, BulkDeleteView):
class AgentManagedDeviceDeleteView(AgentSecurityMixin, PageContextMixin, BulkDeleteView):
"""Confirm and bulk-delete AGENT_MANAGED_DEVICE records."""

template_name = 'agents/targets/confirm_delete.html'
Expand Down Expand Up @@ -472,7 +481,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]
Expand All @@ -498,7 +509,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]
Expand Down Expand Up @@ -541,7 +554,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)."""

Expand Down Expand Up @@ -579,7 +592,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
Expand Down Expand Up @@ -624,7 +637,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]
Expand Down
72 changes: 66 additions & 6 deletions trustpoint/devices/forms.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Comment on lines +75 to +79
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.

Expand Down Expand Up @@ -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

Expand Down Expand Up @@ -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
Expand All @@ -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
Expand Down
4 changes: 2 additions & 2 deletions trustpoint/management/models/security.py
Original file line number Diff line number Diff line change
Expand Up @@ -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]
Comment on lines 216 to +219

_MODE_DEFAULTS: ClassVar[dict[str, _SecurityModeDefaults]] = {
# ----------------------------------------------------------------
Expand Down
Loading