Skip to content
Merged
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
1 change: 1 addition & 0 deletions .env.example
Original file line number Diff line number Diff line change
Expand Up @@ -128,6 +128,7 @@ TP_SECURITY_MODE=LAB
# TP_SECURITY_RSA_MINIMUM_KEY_SIZE=2048
# TP_SECURITY_MAX_CERT_VALIDITY_DAYS=365
# TP_SECURITY_MAX_CRL_VALIDITY_DAYS=90
# Default lifetime for CMP, EST, and REST onboarding credentials, in seconds (null for no expiry).
# TP_SECURITY_ALLOW_CA_ISSUANCE=false
# TP_SECURITY_ALLOW_AUTO_GEN_PKI=false
# TP_SECURITY_ALLOW_SELF_SIGNED_CA=false
Expand Down
44 changes: 44 additions & 0 deletions trustpoint/devices/serializers.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,8 +8,10 @@
"""

import secrets
from datetime import timedelta
from typing import Any, ClassVar

from django.utils import timezone
from rest_framework import serializers

from onboarding.models import OnboardingConfigModel, OnboardingPkiProtocol, OnboardingProtocol
Expand Down Expand Up @@ -53,6 +55,18 @@ class OnboardingConfigSerializer(serializers.ModelSerializer[OnboardingConfigMod
)
)

credential_ttl_seconds = serializers.IntegerField(
required=False,
write_only=True,
min_value=1,
help_text=(
'Optional validity window (in seconds) for CMP, EST, and REST credentials. '
'When provided, credential_expires_at is set to now + ttl. '
'If omitted, new credentials use the TTL from Security Configuration '
'unless an expiry is explicitly given.'
),
)

class Meta:
"""Metadata for OnboardingConfigSerializer."""

Expand All @@ -64,6 +78,8 @@ class Meta:
'pki_protocols',
'est_password',
'cmp_shared_secret',
'credential_expires_at',
'credential_ttl_seconds',
'opc_user',
'opc_password',
'idevid_trust_store',
Expand All @@ -87,6 +103,25 @@ class Meta:
},
}

def _apply_credential_expiry(
self,
data: dict[str, Any],
ttl_seconds: int | None,
) -> None:
"""Resolve the credential expiry into data['credential_expires_at'].

Precedence:
1. An explicit credential_expires_at already in data is left untouched.
2. An explicit ttl_seconds sets expiry to now + ttl.
3. The model applies the security configuration default when saving a new credential.
"""
if data.get('credential_expires_at') is not None:
return

if ttl_seconds is not None:
data['credential_expires_at'] = timezone.now() + timedelta(seconds=ttl_seconds)
return

def _generate_secure_secret(self, length: int = 32) -> str:
"""Generate a cryptographically secure random secret.

Expand Down Expand Up @@ -138,6 +173,7 @@ def create(self, validated_data: dict[str, Any]) -> OnboardingConfigModel:
- EST_USERNAME_PASSWORD or REST_USERNAME_PASSWORD: generates est_password
"""
pki_protocol_values = validated_data.pop('pki_protocols', [])
ttl_seconds = validated_data.pop('credential_ttl_seconds', None)
onboarding_protocol = validated_data.get('onboarding_protocol')

if onboarding_protocol == OnboardingProtocol.CMP_SHARED_SECRET and not validated_data.get('cmp_shared_secret'):
Expand All @@ -147,6 +183,8 @@ def create(self, validated_data: dict[str, Any]) -> OnboardingConfigModel:
and not validated_data.get('est_password')):
validated_data['est_password'] = self._generate_secure_secret()

self._apply_credential_expiry(validated_data, ttl_seconds)

instance = OnboardingConfigModel(**validated_data)

if pki_protocol_values:
Expand All @@ -159,6 +197,12 @@ def create(self, validated_data: dict[str, Any]) -> OnboardingConfigModel:
def update(self, instance: OnboardingConfigModel, validated_data: dict[str, Any]) -> OnboardingConfigModel:
"""Update OnboardingConfigModel instance with proper PKI protocol handling."""
pki_protocol_values = validated_data.pop('pki_protocols', None)
ttl_seconds = validated_data.pop('credential_ttl_seconds', None)

# Resolve an explicit TTL into an expiry timestamp. A TTL always takes
# precedence over any expiry already present in the update payload.
if ttl_seconds is not None:
validated_data['credential_expires_at'] = timezone.now() + timedelta(seconds=ttl_seconds)

for attr, value in validated_data.items():
setattr(instance, attr, value)
Expand Down
120 changes: 120 additions & 0 deletions trustpoint/devices/tests/test_api/test_credential_expiry.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,120 @@
# Copyright (c) 2026 The Trustpoint Project Authors
# SPDX-License-Identifier: MIT

"""Tests for protocol-independent credential expiry API inputs."""

from datetime import datetime, timedelta, timezone
from unittest.mock import patch

import pytest

from devices.serializers import OnboardingConfigSerializer
from management.models import SecurityConfig
from onboarding.models import OnboardingConfigModel, OnboardingProtocol


@pytest.mark.django_db
@pytest.mark.parametrize('protocol', [
OnboardingProtocol.CMP_SHARED_SECRET,
OnboardingProtocol.EST_USERNAME_PASSWORD,
OnboardingProtocol.REST_USERNAME_PASSWORD,
])
def test_credential_ttl_create_and_update(protocol: OnboardingProtocol) -> None:
"""The same TTL input applies to CMP, EST, and REST and survives database storage."""
now = datetime(2026, 10, 2, tzinfo=timezone.utc)
serializer = OnboardingConfigSerializer(data={
'onboarding_protocol': protocol,
'credential_ttl_seconds': 120,
})
assert serializer.is_valid(), serializer.errors
with patch('devices.serializers.timezone.now', return_value=now):
config = serializer.save()
config.refresh_from_db()
assert config.credential_expires_at == now + timedelta(seconds=120)
assert 'credential_ttl_seconds' not in serializer.data

update = OnboardingConfigSerializer(config, data={'credential_ttl_seconds': 240}, partial=True)
assert update.is_valid(), update.errors
with patch('devices.serializers.timezone.now', return_value=now):
update.save()
config.refresh_from_db()
assert config.credential_expires_at == now + timedelta(seconds=240)


@pytest.mark.django_db
@pytest.mark.parametrize('protocol', [
OnboardingProtocol.CMP_SHARED_SECRET,
OnboardingProtocol.EST_USERNAME_PASSWORD,
OnboardingProtocol.REST_USERNAME_PASSWORD,
])
def test_explicit_credential_expiry(protocol: OnboardingProtocol) -> None:
"""All protocols accept an explicit timestamp and allow clearing the expiry."""
expiry = datetime(2026, 10, 2, tzinfo=timezone.utc)
serializer = OnboardingConfigSerializer(data={
'onboarding_protocol': protocol,
'credential_expires_at': expiry.isoformat(),
})
assert serializer.is_valid(), serializer.errors
config = serializer.save()
config.refresh_from_db()
assert config.credential_expires_at == expiry
assert 'credential_expires_at' in serializer.data

update = OnboardingConfigSerializer(config, data={'credential_expires_at': None}, partial=True)
assert update.is_valid(), update.errors
update.save()
config.refresh_from_db()
assert config.credential_expires_at is None


@pytest.mark.django_db
@pytest.mark.parametrize('mode', list(SecurityConfig.SecurityModeChoices))
@pytest.mark.parametrize('protocol', [
OnboardingProtocol.CMP_SHARED_SECRET,
OnboardingProtocol.EST_USERNAME_PASSWORD,
OnboardingProtocol.REST_USERNAME_PASSWORD,
])
def test_security_mode_credential_ttl(mode: str, protocol: OnboardingProtocol) -> None:
"""Both API and direct model creation use the configured mode's credential TTL."""
security = SecurityConfig.objects.create(pk=1, security_mode=mode)
security.apply_security_settings()
now = datetime(2026, 10, 2, tzinfo=timezone.utc)
expected = now + timedelta(seconds=600) if mode in (
SecurityConfig.SecurityModeChoices.HARDENED,
SecurityConfig.SecurityModeChoices.CRITICAL,
) else None
serializer = OnboardingConfigSerializer(data={'onboarding_protocol': protocol})
assert serializer.is_valid(), serializer.errors
credential_field = 'cmp_shared_secret' if protocol == OnboardingProtocol.CMP_SHARED_SECRET else 'est_password'
with patch('onboarding.models.timezone.now', return_value=now):
api_config = serializer.save()
model_config = OnboardingConfigModel.objects.create(
onboarding_protocol=protocol, **{credential_field: 'test-secret'},
)
for config in (api_config, model_config):
config.refresh_from_db()
assert config.credential_expires_at == expected
with patch('onboarding.models.timezone.now', return_value=now + timedelta(seconds=900)):
config.save()
config.refresh_from_db()
assert config.credential_expires_at == expected


@pytest.mark.django_db
@pytest.mark.parametrize('ttl_seconds', [None, 300])
@pytest.mark.parametrize('protocol', [
OnboardingProtocol.CMP_SHARED_SECRET,
OnboardingProtocol.EST_USERNAME_PASSWORD,
OnboardingProtocol.REST_USERNAME_PASSWORD,
])
def test_custom_security_credential_ttl(ttl_seconds: int | None, protocol: OnboardingProtocol) -> None:
"""A custom security TTL applies to all three shared-credential protocols."""
SecurityConfig.objects.create(pk=1, credential_ttl_seconds=ttl_seconds)
now = datetime(2026, 10, 2, tzinfo=timezone.utc)
serializer = OnboardingConfigSerializer(data={'onboarding_protocol': protocol})
assert serializer.is_valid(), serializer.errors
with patch('onboarding.models.timezone.now', return_value=now):
config = serializer.save()
assert config.credential_expires_at == (
None if ttl_seconds is None else now + timedelta(seconds=ttl_seconds)
)
2 changes: 1 addition & 1 deletion trustpoint/features/steps/common_steps.py
Original file line number Diff line number Diff line change
Expand Up @@ -234,7 +234,7 @@ def step_security_option_configure(context: runner.Context, opt_name: str, opt_v

assert SecurityConfig.objects.filter(id=1).exists(), 'SecurityConfig does not exist'
form = response.context['security_form']
data = form.initial.copy()
data = {name: '' if value is None else value for name, value in form.initial.items()}

data['max_cert_validity_days'] = 1825
data['max_crl_validity_days'] = 365
Expand Down
59 changes: 58 additions & 1 deletion trustpoint/help_pages/devices_help_views.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,13 +7,17 @@

import ipaddress
import json
import secrets
from datetime import timedelta
from pathlib import Path
from typing import TYPE_CHECKING, Any, ClassVar, override

from cryptography import x509
from django.conf import settings
from django.contrib import messages
from django.core.exceptions import PermissionDenied
from django.core.management import call_command
from django.db import transaction
from django.http import FileResponse, Http404, HttpResponse, HttpResponseRedirect
from django.urls import reverse
from django.utils import timezone
Expand Down Expand Up @@ -47,6 +51,8 @@
)
from help_pages.forms import IpAddressForm
from help_pages.help_section import HelpPage, HelpRow, HelpSection, ValueRenderType
from management.models import SecurityConfig
from onboarding.models import OnboardingConfigModel, OnboardingProtocol
from pki.models import IssuedCredentialModel
from pki.models.certificate import RevokedCertificateModel
from pki.models.truststore import ActiveTrustpointTlsServerCredentialModel
Expand All @@ -65,6 +71,8 @@
if TYPE_CHECKING:
from typing import Any

from django.http import HttpRequest

from pki.models import CaModel, CredentialModel


Expand All @@ -75,7 +83,7 @@ class BaseHelpView(UserPermissionRequiredMixin, PageContextMixin, DetailView[Dev
"""Base help view that constructs the context."""

template_name = 'help/help_page.html'
http_method_names = ('get',)
http_method_names = ('get', 'post')
model = DeviceModel
context_object_name = 'device'
permission_required = AppPermissions.VIEW_HELP_PAGES
Expand All @@ -84,6 +92,48 @@ class BaseHelpView(UserPermissionRequiredMixin, PageContextMixin, DetailView[Dev
page_name: str
strategy: HelpPageStrategy

def uses_onboarding_credential(self) -> bool:
"""Return whether this help page uses an onboarding password or shared secret."""
return isinstance(self.strategy, (
OnboardingDomainCredentialCmpSharedSecretStrategy,
OnboardingDomainCredentialEstUsernamePasswordStrategy,
OnboardingDomainCredentialRestUsernamePasswordStrategy,
AgentSetupProfileStrategy,
))

def post(self, request: HttpRequest, *args: Any, **kwargs: Any) -> HttpResponseRedirect:
"""Rotate an expired onboarding credential and restart its configured lifetime."""
del args, kwargs
if not request.user.has_perm(AppPermissions.MANAGE_DEVICES):
raise PermissionDenied
self.object = self.get_object()
if not self.uses_onboarding_credential() or not self.object.onboarding_config_id:
raise Http404(_('No renewable onboarding credential is configured for this device.'))
with transaction.atomic():
config = OnboardingConfigModel.objects.select_for_update().get(pk=self.object.onboarding_config_id)
if config.onboarding_protocol == OnboardingProtocol.CMP_SHARED_SECRET:
credential_field = 'cmp_shared_secret'
elif config.onboarding_protocol in (
OnboardingProtocol.EST_USERNAME_PASSWORD,
OnboardingProtocol.REST_USERNAME_PASSWORD,
OnboardingProtocol.AGENT,
):
credential_field = 'est_password'
else:
raise Http404(_('No renewable onboarding credential is configured for this device.'))
if config.is_credential_expired() and getattr(config, credential_field):
security_config = SecurityConfig.objects.filter(pk=1).first()
ttl_seconds = security_config.credential_ttl_seconds if security_config is not None else None
setattr(config, credential_field, secrets.token_urlsafe(32))
config.credential_expires_at = (
timezone.now() + timedelta(seconds=ttl_seconds) if ttl_seconds is not None else None
)
config.save(update_fields=[credential_field, 'credential_expires_at'])
messages.success(request, _('The onboarding credential was renewed.'))
else:
messages.warning(request, _('The onboarding credential is not expired or is no longer available.'))
return HttpResponseRedirect(request.get_full_path())

def _make_context(self, host_ip: str = '127.0.0.1') -> HelpContext:
device = self.object
domain = getattr(device, 'domain', None)
Expand Down Expand Up @@ -165,6 +215,13 @@ def get_context_data(self, **kwargs: Any) -> dict[str, Any]:

sections, heading = self.strategy.build_sections(help_context=help_context)

onboarding_config = self.object.onboarding_config
context['credential_expired'] = bool(
self.uses_onboarding_credential()
and onboarding_config
and onboarding_config.is_credential_expired()
)
context['can_renew_credential'] = self.request.user.has_perm(AppPermissions.MANAGE_DEVICES)
context['help_page'] = HelpPage(heading=heading, sections=sections)
context['ValueRenderType_CODE'] = ValueRenderType.CODE.value
context['ValueRenderType_PLAIN'] = ValueRenderType.PLAIN.value
Expand Down
Loading
Loading