"""This module contains all views concerning the devices application."""
import abc
import asyncio
import datetime
import io
import json
import zipfile
from collections.abc import Sequence
from typing import Any, ClassVar, cast
from cryptography import x509
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import hashes, serialization
from django import forms
from django.contrib import messages
from django.contrib.auth.decorators import login_not_required
from django.core.paginator import Paginator
from django.db.models import Q, QuerySet
from django.http import FileResponse, Http404, HttpResponse, HttpResponseBase, HttpResponseRedirect
from django.http.request import HttpRequest
from django.shortcuts import get_object_or_404, redirect
from django.urls import reverse, reverse_lazy
from django.utils.decorators import method_decorator
from django.utils.html import format_html
from django.utils.safestring import SafeString
from django.utils.translation import gettext_lazy, ngettext
from django.views.generic.base import RedirectView, TemplateView
from django.views.generic.detail import DetailView
from django.views.generic.edit import FormMixin, FormView
from django.views.generic.list import ListView
from drf_spectacular.utils import extend_schema
from rest_framework import viewsets
from trustpoint_core.archiver import Archiver
from trustpoint_core.serializer import CredentialFileFormat
from devices.filters import DeviceFilter
from devices.forms import (
BaseCredentialForm,
BrowserLoginForm,
ClmDeviceModelNoOnboardingForm,
ClmDeviceModelOnboardingForm,
ClmDeviceModelOpcUaGdsPushOnboardingForm,
CredentialDownloadForm,
DeleteDevicesForm,
IssueDomainCredentialForm,
IssueOpcUaGdsPushDomainCredentialForm,
NoOnboardingCreateForm,
OnboardingCreateForm,
OpcUaGdsPushCreateForm,
OpcUaGdsPushTruststoreAssociationForm,
RevokeDevicesForm,
RevokeIssuedCredentialForm,
)
# noinspection PyUnresolvedReferences
from devices.issuer import (
BaseTlsCredentialIssuer,
LocalDomainCredentialIssuer,
)
from devices.models import (
DeviceModel,
RemoteDeviceCredentialDownloadModel,
)
from devices.revocation import DeviceCredentialRevocation
from devices.serializers import DeviceSerializer
from management.models import KeyStorageConfig
from onboarding.models import (
NoOnboardingPkiProtocol,
OnboardingPkiProtocol,
OnboardingProtocol,
OnboardingStatus,
)
from pki.forms import TruststoreAddForm
from pki.forms.cert_profiles import CertificateIssuanceForm
from pki.models import IssuedCredentialModel, RemoteIssuedCredentialModel
from pki.models.ca import CaModel
from pki.models.cert_profile import CertificateProfileModel
from pki.models.certificate import CertificateModel, RevokedCertificateModel
from pki.models.credential import CredentialModel, OwnerCredentialModel
from pki.models.domain import DomainAllowedCertificateProfileModel
from pki.models.truststore import TruststoreModel, TruststoreOrderModel
from request.authorization import ManualAuthorization
from request.gds_push import GdsPushService
from request.gds_push.gds_push_service import GdsPushError
from request.operation_processor.issue_cred import CredentialIssueProcessor
from request.request_context import ManualCredentialRequestContext
from trustpoint.logger import LoggerMixin
from trustpoint.page_context import (
DEVICES_PAGE_CATEGORY,
DEVICES_PAGE_DEVICES_SUBCATEGORY,
DEVICES_PAGE_OPC_UA_SUBCATEGORY,
PageContextMixin,
)
from trustpoint.settings import UIConfig
from util.mult_obj_views import get_primary_keys_from_str_as_list_of_ints
[docs]
DeviceWithoutDomainErrorMsg = gettext_lazy('Device does not have an associated domain.')
[docs]
NamedCurveMissingForEccErrorMsg = gettext_lazy('Failed to retrieve named curve for ECC algorithm.')
[docs]
ActiveTrustpointTlsServerCredentialModelMissingErrorMsg = gettext_lazy(
'No active trustpoint TLS server credential found.'
)
# This only occurs if no domain is configured
[docs]
PublicKeyInfoMissingErrorMsg = DeviceWithoutDomainErrorMsg
# -------------------------------------------------- Main Table Views --------------------------------------------------
[docs]
class AbstractDeviceTableView(PageContextMixin, ListView[DeviceModel], abc.ABC):
"""Device Table View."""
[docs]
http_method_names = ('get',)
[docs]
context_object_name = 'devices'
[docs]
paginate_by = UIConfig.paginate_by
[docs]
default_sort_param = 'common_name'
[docs]
filterset_class = DeviceFilter
[docs]
page_category = DEVICES_PAGE_CATEGORY
[docs]
def apply_filters(self, qs: QuerySet[DeviceModel]) -> QuerySet[DeviceModel]:
"""Applies the `DeviceFilter` to the given queryset.
Args:
qs: The base queryset to filter.
Returns:
The filtered queryset according to GET parameters.
"""
self.filterset = DeviceFilter(self.request.GET, queryset=qs)
return cast('QuerySet[DeviceModel]', self.filterset.qs)
[docs]
def get(self, request: HttpRequest, *args: Any, **kwargs: Any) -> HttpResponse:
"""Adds the object model to the instance and forwards to super().get().
Args:
request: The Django request object.
*args: Positional arguments passed to super().get().
**kwargs: Keyword arguments passed to super().get().
Returns:
The HttpResponse object returned by super().get().
"""
sort_params = request.GET.getlist('sort', [self.default_sort_param])
if len(sort_params) > 1:
first_sort_parameter = sort_params[0]
query_dict = request.GET.copy()
query_dict.setlist('sort', [first_sort_parameter])
new_url = f'{request.path}?{query_dict.urlencode()}'
return HttpResponseRedirect(new_url)
self.ordering = sort_params[0]
return super().get(request, *args, **kwargs)
@abc.abstractmethod
[docs]
def get_queryset(self) -> QuerySet[DeviceModel]:
"""Filter queryset to only include devices which are of generic type.
Returns:
Returns a queryset of all DeviceModels which are of generic type.
"""
...
[docs]
def get_context_data(self, **kwargs: Any) -> dict[str, Any]:
"""Adds the clm and revoke buttons to the context.
Args:
**kwargs: Keyword arguments passed to super().get_context_data.
Returns:
The context to use for rendering the devices page.
"""
context = super().get_context_data(**kwargs)
sort_param = self.request.GET.get('sort', self.default_sort_param)
context['current_sort'] = sort_param
context['filter'] = getattr(self, 'filterset', None)
params = self.request.GET.copy()
params.pop('sort', None)
context['preserve_qs'] = params.urlencode()
for device in context['devices']:
device.clm_button = self._get_clm_button_html(device)
device.pki_protocols = self._get_pki_protocols(device)
context['create_url'] = f'{self.page_category}:{self.page_name}_create'
context['new_onboarding_url'] = f'{self.page_category}:{self.page_name}_new_onboarding'
context['device_revoke_url'] = reverse(f'{self.page_category}:{self.page_name}_device_revoke')
context['device_delete_url'] = reverse(f'{self.page_category}:{self.page_name}_device_delete')
return context
[docs]
def get_ordering(self) -> str | Sequence[str] | None:
"""Returns the sort parameters as a list.
Returns:
The sort parameters, if any. Otherwise the default sort parameter.
"""
return self.request.GET.getlist('sort', [self.default_sort_param])
def _get_clm_button_html(self, record: DeviceModel) -> SafeString:
"""Gets the HTML for the CLM button in the devices table.
Args:
record: The corresponding DeviceModel.
Returns:
The HTML of the hyperlink for the CLM button.
"""
if record.device_type == DeviceModel.DeviceType.OPC_UA_GDS_PUSH:
clm_url = reverse('devices:opc_ua_gds_push_certificate_lifecycle_management', kwargs={'pk': record.pk})
else:
clm_url = reverse(
f'{self.page_category}:{self.page_name}_certificate_lifecycle_management', kwargs={'pk': record.pk}
)
return format_html(
'<a href="{}" class="btn btn-primary tp-table-btn w-100">{}</a>', clm_url, gettext_lazy('Manage')
)
def _get_pki_protocols(self, record: DeviceModel) -> str:
if record.onboarding_config:
return ', '.join([str(p.label) for p in record.onboarding_config.get_pki_protocols()])
if record.no_onboarding_config:
return ', '.join([str(p.label) for p in record.no_onboarding_config.get_pki_protocols()])
return ''
[docs]
class DeviceTableView(AbstractDeviceTableView):
"""Device Table View."""
[docs]
template_name = 'devices/devices.html'
[docs]
page_name = DEVICES_PAGE_DEVICES_SUBCATEGORY
[docs]
def get_queryset(self) -> QuerySet[DeviceModel]:
"""Filter queryset to include all device types (Generic, OPC UA GDS Push) and filtered by UI filters.
Returns:
Returns a queryset of all DeviceModels (excluding OPC UA GDS), filtered by UI filters.
"""
base_qs = super(ListView, self).get_queryset().exclude(
device_type=DeviceModel.DeviceType.OPC_UA_GDS
)
return self.apply_filters(base_qs)
[docs]
class OpcUaGdsTableView(DeviceTableView):
"""Table View for devices where opc_ua_gds is True."""
[docs]
template_name = 'devices/opc_ua_gds.html'
[docs]
page_name = DEVICES_PAGE_OPC_UA_SUBCATEGORY
[docs]
def get_queryset(self) -> QuerySet[DeviceModel]:
"""Filter queryset to only include devices which are of OPC-UA GDS type and filtered by UI filters.
Returns:
Returns a queryset of all DeviceModels which are of OPC-UA GDS type, filtered by UI filters.
"""
base_qs = super(ListView, self).get_queryset().filter(
device_type=DeviceModel.DeviceType.OPC_UA_GDS
)
return self.apply_filters(base_qs)
# ------------------------------------------------- Device Create View -------------------------------------------------
[docs]
class AbstractCreateChooseOnboaringView(PageContextMixin, TemplateView):
"""Abstract view for choosing if the new device shall be onboarded or not."""
[docs]
http_method_names = ('get',)
[docs]
template_name = 'devices/create_choose_onboarding.html'
[docs]
page_category = DEVICES_PAGE_CATEGORY
[docs]
def get_context_data(self, **kwargs: Any) -> dict[str, Any]:
"""Adds the cancel url href according to the subcategory.
Args:
**kwargs: Keyword arguments passed to super().get_context_data.
Returns:
The context to use for rendering the devices page.
"""
context = super().get_context_data(**kwargs)
context['cancel_create_url'] = f'devices:{self.page_name}'
context['use_onboarding_url'] = f'{self.page_category}:{self.page_name}_create_onboarding'
context['use_no_onboarding_url'] = f'{self.page_category}:{self.page_name}_create_no_onboarding'
context['use_opc_ua_gds_push_url'] = f'{self.page_category}:{self.page_name}_create_opc_ua_gds_push'
context['show_opc_ua_gds_push_option'] = False # Default: don't show GDS Push option
return context
[docs]
class DeviceCreateChooseOnboardingView(AbstractCreateChooseOnboaringView):
"""View for choosing if the new device shall be onboarded or not."""
[docs]
page_name = DEVICES_PAGE_DEVICES_SUBCATEGORY
[docs]
def get_context_data(self, **kwargs: Any) -> dict[str, Any]:
"""Adds the cancel url href according to the subcategory.
Args:
**kwargs: Keyword arguments passed to super().get_context_data.
Returns:
The context to use for rendering the devices page.
"""
context = super().get_context_data(**kwargs)
# Enable OPC UA GDS Push option for standard devices
context['show_opc_ua_gds_push_option'] = True
return context
[docs]
class OpcUaGdsCreateChooseOnboardingView(AbstractCreateChooseOnboaringView):
"""View for choosing if the new OPC UA GDS shall be onboarded or not."""
[docs]
page_name = DEVICES_PAGE_OPC_UA_SUBCATEGORY
[docs]
def get_context_data(self, **kwargs: Any) -> dict[str, Any]:
"""Adds the cancel url href according to the subcategory.
Args:
**kwargs: Keyword arguments passed to super().get_context_data.
Returns:
The context to use for rendering the devices page.
"""
context = super().get_context_data(**kwargs)
# Remove no-onboarding and OPC UA GDS Push options for OPC UA GDS
context.pop('use_no_onboarding_url', None)
context['show_opc_ua_gds_push_option'] = False
return context
[docs]
class OpcUaGdsPushCreateChooseOnboardingView(RedirectView):
"""Deprecated: Redirects to standard devices create page."""
[docs]
pattern_name = 'devices:devices_create'
[docs]
class AbstractCreateAddOnboardingTypeView(PageContextMixin, TemplateView):
"""Abstract view for choosing how new device shall be added."""
[docs]
http_method_names = ('get',)
[docs]
template_name = 'devices/add_onboarding_type.html'
[docs]
page_category = DEVICES_PAGE_CATEGORY
[docs]
def get_context_data(self, **kwargs: Any) -> dict[str, Any]:
"""Adds the cancel url href according to the subcategory.
Args:
**kwargs: Keyword arguments passed to super().get_context_data.
Returns:
The context to use for rendering the devices page.
"""
context = super().get_context_data(**kwargs)
context['cancel_create_url'] = f'devices:{self.page_name}'
context['use_onboarding_url_name'] = 'pki:devid_registration-method_select'
context['use_no_onboarding_url'] = f'{self.page_category}:{self.page_name}_create_no_onboarding'
return context
[docs]
class DeviceCreateAddOnboardingTypeView(AbstractCreateAddOnboardingTypeView):
"""View for choosing how new device shall be added."""
[docs]
page_name = DEVICES_PAGE_DEVICES_SUBCATEGORY
[docs]
class AbstractCreateNoOnboardingView(PageContextMixin, FormView[NoOnboardingCreateForm]):
"""asdfds."""
[docs]
http_method_names = ('get', 'post')
[docs]
template_name = 'devices/create.html'
[docs]
page_category = DEVICES_PAGE_CATEGORY
[docs]
def get_context_data(self, **kwargs: Any) -> dict[str, Any]:
"""Adds the cancel url href according to the subcategory.
Args:
**kwargs: Keyword arguments passed to super().get_context_data.
Returns:
The context to use for rendering the devices page.
"""
context = super().get_context_data(**kwargs)
context['cancel_create_url'] = f'{self.page_category}:{self.page_name}'
return context
[docs]
def get_success_url(self) -> str:
"""Gets the success url to redirect to after successful processing of the POST data following a form submit.
Returns:
The success url to redirect to after successful processing of the POST data following a form submit.
"""
return str(
reverse_lazy(
f'{self.page_category}:{self.page_name}_certificate_lifecycle_management', kwargs={'pk': self.object.id}
)
)
[docs]
class DeviceCreateNoOnboardingView(AbstractCreateNoOnboardingView):
"""Create form view for the devices section."""
[docs]
page_name = DEVICES_PAGE_DEVICES_SUBCATEGORY
[docs]
class OpcUaGdsCreateNoOnboardingView(AbstractCreateNoOnboardingView):
"""Create form view for the devices section."""
[docs]
page_name = DEVICES_PAGE_OPC_UA_SUBCATEGORY
[docs]
class AbstractCreateOnboardingView(PageContextMixin, FormView[forms.Form]):
"""asdfds."""
[docs]
http_method_names = ('get', 'post')
[docs]
template_name = 'devices/create.html'
[docs]
page_category = DEVICES_PAGE_CATEGORY
[docs]
def get_context_data(self, **kwargs: Any) -> dict[str, Any]:
"""Adds the cancel url href according to the subcategory.
Args:
**kwargs: Keyword arguments passed to super().get_context_data.
Returns:
The context to use for rendering the devices page.
"""
context = super().get_context_data(**kwargs)
context['cancel_create_url'] = f'{self.page_category}:{self.page_name}'
return context
[docs]
def get_success_url(self) -> str:
"""Gets the success url to redirect to after successful processing of the POST data following a form submit.
Returns:
The success url to redirect to after successful processing of the POST data following a form submit.
"""
return str(
reverse_lazy(
f'{self.page_category}:{self.page_name}_certificate_lifecycle_management', kwargs={'pk': self.object.id}
)
)
[docs]
class DeviceCreateOnboardingView(AbstractCreateOnboardingView):
"""Create form view for the devices section."""
[docs]
page_name = DEVICES_PAGE_DEVICES_SUBCATEGORY
[docs]
class DeviceCreateOpcUaGdsPushView(AbstractCreateOnboardingView):
"""Create form view for OPC UA GDS Push devices in the devices section."""
[docs]
page_name = DEVICES_PAGE_DEVICES_SUBCATEGORY
[docs]
def dispatch(self, request: HttpRequest, *args: Any, **kwargs: Any) -> HttpResponseBase:
"""Check if SOFTWARE storage is configured before allowing GDS Push creation.
Args:
request: The HTTP request object.
*args: Variable length argument list.
**kwargs: Arbitrary keyword arguments.
Returns:
HttpResponseBase: The response object.
"""
try:
config = KeyStorageConfig.get_config()
if config.storage_type != KeyStorageConfig.StorageType.SOFTWARE:
messages.error(
request,
f'OPC UA GDS Push is only available with SOFTWARE key storage. '
f'Current storage type: {config.get_storage_type_display()}'
)
return redirect('devices:devices')
except KeyStorageConfig.DoesNotExist:
messages.error(
request,
'Key storage configuration not found. Please configure key storage first.'
)
return redirect('devices:devices')
return super().dispatch(request, *args, **kwargs)
[docs]
class OpcUaGdsCreateOnboardingView(AbstractCreateOnboardingView):
"""Create form view for the devices section."""
[docs]
page_name = DEVICES_PAGE_OPC_UA_SUBCATEGORY
[docs]
class OpcUaGdsPushCreateOnboardingView(RedirectView):
"""Deprecated: Redirects to standard devices OPC UA GDS Push create page."""
[docs]
pattern_name = 'devices:devices_create_opc_ua_gds_push'
# ------------------------------------------ Certificate Lifecycle Management ------------------------------------------
[docs]
class AbstractCertificateLifecycleManagementSummaryView(PageContextMixin, DetailView[DeviceModel], abc.ABC):
"""This is the CLM summary view in the devices section."""
[docs]
http_method_names = ('get', 'post')
[docs]
template_name = 'devices/credentials/certificate_lifecycle_management.html'
[docs]
context_object_name = 'device'
[docs]
default_sort_param = 'common_name'
[docs]
issued_creds_qs: QuerySet[IssuedCredentialModel]
[docs]
remote_issued_creds_qs: QuerySet[RemoteIssuedCredentialModel]
[docs]
domain_credentials_qs: QuerySet[IssuedCredentialModel] | QuerySet[RemoteIssuedCredentialModel]
[docs]
application_credentials_qs: QuerySet[IssuedCredentialModel] | QuerySet[RemoteIssuedCredentialModel]
[docs]
page_category = DEVICES_PAGE_CATEGORY
_RA_CA_TYPES = (CaModel.CaTypeChoice.REMOTE_EST_RA, CaModel.CaTypeChoice.REMOTE_CMP_RA)
def _is_ra_domain(self) -> bool:
"""Return True if the device's domain uses a remote RA CA (EST or CMP)."""
device = self.object
if not device.domain or not device.domain.issuing_ca:
return False
return device.domain.issuing_ca.ca_type in self._RA_CA_TYPES
def _get_owner_credential_for_device(self) -> OwnerCredentialModel | None:
"""Find the OwnerCredentialModel associated with this device."""
device_domain_cred = IssuedCredentialModel.objects.filter(
device=self.object,
issued_credential_type=IssuedCredentialModel.IssuedCredentialType.DOMAIN_CREDENTIAL,
).select_related('credential__certificate').first()
if device_domain_cred is None or device_domain_cred.credential.certificate is None:
return None
fingerprint = device_domain_cred.credential.certificate.sha256_fingerprint
owner_ic = RemoteIssuedCredentialModel.objects.filter(
credential__certificates__sha256_fingerprint=fingerprint,
owner_credential__isnull=False,
issued_credential_type=RemoteIssuedCredentialModel.RemoteIssuedCredentialType.DOMAIN_CREDENTIAL,
).select_related('owner_credential').first()
if owner_ic is None:
return None
return owner_ic.owner_credential
def _get_dev_owner_id_credentials_qs(
self, owner_credential: OwnerCredentialModel,
) -> QuerySet[RemoteIssuedCredentialModel]:
"""Return enrolled DevOwnerID credentials for the given owner credential."""
return RemoteIssuedCredentialModel.objects.filter(
owner_credential=owner_credential,
issued_credential_type=RemoteIssuedCredentialModel.RemoteIssuedCredentialType.DEV_OWNER_ID,
credential__certificate__isnull=False,
).select_related('credential__certificate').order_by('-created_at')
[docs]
def get_issued_creds_qs(self) -> QuerySet[IssuedCredentialModel]:
"""Gets a sorted queryset of all IssuedCredentialModels.
Returns:
Sorted queryset of all IssuedCredentialModels.
"""
issued_creds_qs = IssuedCredentialModel.objects.all()
sort_param = self.request.GET.get('sort', self.default_sort_param)
return issued_creds_qs.order_by(sort_param)
[docs]
def get_remote_issued_creds_qs(self) -> QuerySet[RemoteIssuedCredentialModel]:
"""Gets a sorted queryset of all RemoteIssuedCredentialModels for RA-issued certs."""
sort_param = self.request.GET.get('sort', self.default_sort_param)
return RemoteIssuedCredentialModel.objects.filter(
device=self.object,
issued_credential_type=RemoteIssuedCredentialModel.RemoteIssuedCredentialType.RA_DEVICE,
).select_related('credential__certificate', 'domain').order_by(sort_param)
[docs]
def get_domain_credentials_qs(
self,
) -> QuerySet[IssuedCredentialModel] | QuerySet[RemoteIssuedCredentialModel]:
"""Gets domain credentials — from RemoteIssuedCredentialModel for RA domains, otherwise IssuedCredentialModel.
self.get_issued_creds_qs() or self.get_remote_issued_creds_qs() must be called first!
Returns:
Sorted queryset of domain credentials.
"""
if self._is_ra_domain():
return self.remote_issued_creds_qs.none()
return self.issued_creds_qs.filter(
Q(device=self.object)
& Q(issued_credential_type=IssuedCredentialModel.IssuedCredentialType.DOMAIN_CREDENTIAL.value)
)
[docs]
def get_application_credentials_qs(
self,
) -> QuerySet[IssuedCredentialModel] | QuerySet[RemoteIssuedCredentialModel]:
"""Gets application credentials — from RemoteIssuedCredentialModel for RA domains.
self.get_issued_creds_qs() or self.get_remote_issued_creds_qs() must be called first!
Returns:
Sorted queryset of application credentials.
"""
if self._is_ra_domain():
return self.remote_issued_creds_qs
return self.issued_creds_qs.filter(
Q(device=self.object)
& Q(issued_credential_type=IssuedCredentialModel.IssuedCredentialType.APPLICATION_CREDENTIAL.value)
)
def _add_credential_pages_to_context(self, context: dict[str, Any]) -> None:
"""Paginate domain, application and DevOwnerID credentials into the template context."""
paginator_domain = Paginator(self.domain_credentials_qs, UIConfig.paginate_by)
page_number_domain = self.request.GET.get('page', 1)
context['domain_credentials'] = paginator_domain.get_page(page_number_domain)
context['is_paginated'] = paginator_domain.num_pages > 1
paginator_application = Paginator(self.application_credentials_qs, UIConfig.paginate_by)
page_number_application = self.request.GET.get('page-a', 1)
context['application_credentials'] = paginator_application.get_page(page_number_application)
context['is_paginated_a'] = paginator_application.num_pages > 1
is_ra = self._is_ra_domain()
for cred in context['domain_credentials']:
cred.expires_in = self._get_expires_in(cred)
cred.expiration_date = cast('datetime.datetime', cred.credential.certificate_or_error.not_valid_after)
cred.revoke = self._get_revoke_button_html(cred) if not is_ra else ''
for cred in context['application_credentials']:
cred.expires_in = self._get_expires_in(cred)
cred.expiration_date = cast('datetime.datetime', cred.credential.certificate_or_error.not_valid_after)
cred.revoke = self._get_revoke_button_html(cred) if not is_ra else ''
self._add_dev_owner_id_context(context)
def _add_dev_owner_id_context(self, context: dict[str, Any]) -> None:
"""Add DevOwnerID credentials linked via the owner credential to the context."""
owner_credential = self._get_owner_credential_for_device()
context['owner_credential'] = owner_credential
if owner_credential is None:
context['dev_owner_id_credentials'] = []
context['is_paginated_d'] = False
return
dev_owner_id_qs = self._get_dev_owner_id_credentials_qs(owner_credential)
paginator_doid = Paginator(dev_owner_id_qs, UIConfig.paginate_by)
page_number_doid = self.request.GET.get('page-d', 1)
context['dev_owner_id_credentials'] = paginator_doid.get_page(page_number_doid)
context['is_paginated_d'] = paginator_doid.num_pages > 1
for cred in context['dev_owner_id_credentials']:
cred.expires_in = self._get_expires_in(cred)
cred.expiration_date = cast(
'datetime.datetime', cred.credential.certificate_or_error.not_valid_after,
)
[docs]
def get_context_data(self, **kwargs: Any) -> dict[str, Any]:
"""Adds the paginator and credential details to the context."""
self.issued_creds_qs = self.get_issued_creds_qs()
self.remote_issued_creds_qs = self.get_remote_issued_creds_qs()
self.domain_credentials_qs = self.get_domain_credentials_qs()
self.application_credentials_qs = self.get_application_credentials_qs()
context = super().get_context_data(**kwargs)
context['domain_credentials'] = self.domain_credentials_qs
context['application_credentials'] = self.application_credentials_qs
self._add_credential_pages_to_context(context)
context['main_url'] = f'{self.page_category}:{self.page_name}'
is_cmp_ra = (
self.object.domain
and self.object.domain.issuing_ca
and self.object.domain.issuing_ca.ca_type == CaModel.CaTypeChoice.REMOTE_CMP_RA
)
context['is_cmp_ra_domain'] = bool(is_cmp_ra)
context['issue_app_cred_no_onboarding_url'] = ''
if (
not is_cmp_ra
and self.object.domain
and self.object.no_onboarding_config
and self.object.no_onboarding_config.get_pki_protocols()
):
context['issue_app_cred_no_onboarding_url'] = (
f'{self.page_category}:{self.page_name}_no_onboarding_clm_issue_application_credential'
)
issue_domain_cred_onboarding_url = ''
if self.object.onboarding_config:
if self.object.onboarding_config.onboarding_protocol == OnboardingProtocol.CMP_SHARED_SECRET:
issue_domain_cred_onboarding_url = (
f'{self.page_category}:{self.page_name}'
'_certificate_lifecycle_management_issue_domain_credential_cmp_shared_secret'
)
elif self.object.onboarding_config.onboarding_protocol == OnboardingProtocol.EST_USERNAME_PASSWORD:
issue_domain_cred_onboarding_url = (
f'{self.page_category}:{self.page_name}'
'_certificate_lifecycle_management_issue_domain_credential_est_username_password'
)
elif self.object.onboarding_config.onboarding_protocol == OnboardingProtocol.OPC_GDS_PUSH:
issue_domain_cred_onboarding_url = (
f'{self.page_category}:{self.page_name}_onboarding_clm_issue_domain_credential'
)
context['issue_app_cred_onboarding_url'] = ''
if self.object.domain and self.object.onboarding_config and self.object.onboarding_config.get_pki_protocols():
context['issue_app_cred_onboarding_url'] = (
f'{self.page_category}:{self.page_name}_onboarding_clm_issue_application_credential'
)
context['issue_domain_cred_onboarding_url'] = issue_domain_cred_onboarding_url
context['download_url'] = f'{self.page_category}:{self.page_name}_download'
context['help_dispatch_domain_url'] = f'{self.page_category}:{self.page_name}_help_dispatch_domain'
context['help_dispatch_device_type_url'] = f'{self.page_category}:{self.page_name}_help_dispatch_domain'
context['pki_protocols'] = self._get_pki_protocols(self.object)
context['OnboardingProtocol'] = OnboardingProtocol
context['OnboardingPkiProtocol'] = OnboardingPkiProtocol
context['NoOnboardingPkiProtocol'] = NoOnboardingPkiProtocol
context['OnboardingStatus'] = OnboardingStatus
context['device_form'] = self.get_device_form()
if self.object.onboarding_config:
context['onboarding_form'] = self.get_onboarding_form()
return context
[docs]
def get_onboarding_initial(self) -> dict[str, Any]:
"""Gets the initial values for onboarding.
Returns:
Initial values for onboarding.
"""
if not self.object.onboarding_config:
err_msg = gettext_lazy('The device does not have onboarding configured.')
raise ValueError(err_msg)
return {
'common_name': self.object.common_name,
'serial_number': self.object.serial_number,
'domain': self.object.domain,
'onboarding_protocol': self.object.onboarding_config.onboarding_protocol,
'onboarding_status': OnboardingStatus(self.object.onboarding_config.onboarding_status).label,
'pki_protocol_cmp': self.object.onboarding_config.has_pki_protocol(OnboardingPkiProtocol.CMP),
'pki_protocol_est': self.object.onboarding_config.has_pki_protocol(OnboardingPkiProtocol.EST),
}
[docs]
def get_no_onboarding_initial(self) -> dict[str, Any]:
"""Gets the initial values for no onboarding.
Returns:
Initial values for no onboarding.
"""
if not self.object.no_onboarding_config:
err_msg = gettext_lazy('The object has onboarding configured.')
raise ValueError(err_msg)
return {
'common_name': self.object.common_name,
'serial_number': self.object.serial_number,
'domain': self.object.domain,
'pki_protocol_cmp': self.object.no_onboarding_config.has_pki_protocol(
NoOnboardingPkiProtocol.CMP_SHARED_SECRET
),
'pki_protocol_est': self.object.no_onboarding_config.has_pki_protocol(
NoOnboardingPkiProtocol.EST_USERNAME_PASSWORD
),
'pki_protocol_manual': self.object.no_onboarding_config.has_pki_protocol(NoOnboardingPkiProtocol.MANUAL),
}
@staticmethod
def _get_expires_in(record: IssuedCredentialModel | RemoteIssuedCredentialModel) -> str:
"""Gets the remaining time until the credential expires as human-readable string.
Args:
record: The corresponding IssuedCredentialModel or RemoteIssuedCredentialModel.
Returns:
The remaining time until the credential expires as human-readable string.
"""
cert = record.credential.certificate_or_error
if cert.certificate_status != CertificateModel.CertificateStatus.OK:
return str(cert.certificate_status.label)
now = datetime.datetime.now(datetime.UTC)
expire_timedelta = cert.not_valid_after - now
days = expire_timedelta.days
hours, remainder = divmod(expire_timedelta.seconds, 3600)
minutes, seconds = divmod(remainder, 60)
return f'{days} days, {hours}:{minutes:02d}:{seconds:02d}'
def _get_revoke_button_html(self, record: IssuedCredentialModel | RemoteIssuedCredentialModel) -> str:
"""Gets the HTML for the revoke button in the devices table.
Args:
record: The corresponding IssuedCredentialModel or RemoteIssuedCredentialModel.
Returns:
The HTML of the hyperlink for the revoke button.
"""
if not isinstance(record, IssuedCredentialModel):
return ''
cert = record.credential.certificate_or_error
if cert.certificate_status == CertificateModel.CertificateStatus.REVOKED:
return format_html('<a class="btn btn-danger tp-table-btn w-100 disabled">{}</a>', gettext_lazy('Revoked'))
url = reverse(f'{self.page_category}:{self.page_name}_credential_revoke', kwargs={'pk': record.pk})
return format_html('<a href="{}" class="btn btn-danger tp-table-btn w-100">{}</a>', url, gettext_lazy('Revoke'))
def _get_pki_protocols(self, record: DeviceModel) -> str:
if record.onboarding_config:
return ', '.join([str(p.label) for p in record.onboarding_config.get_pki_protocols()])
if record.no_onboarding_config:
return ', '.join([str(p.label) for p in record.no_onboarding_config.get_pki_protocols()])
return ''
[docs]
def post(self, request: HttpRequest, *_args: Any, **kwargs: Any) -> HttpResponse:
"""Handles the POST request used for device form submission.
Args:
request: The django request object.
_args: Positional arguments are discarded.
kwargs: Keyword arguments are passed to get_context_data.
Returns:
The HttpResponse.
"""
self.object = self.get_object()
form: ClmDeviceModelOnboardingForm | ClmDeviceModelNoOnboardingForm
if self.object.onboarding_config:
form = ClmDeviceModelOnboardingForm(request.POST, instance=self.object)
if form.is_valid():
form.save(onboarding_protocol=OnboardingProtocol(self.object.onboarding_config.onboarding_protocol))
else:
form = ClmDeviceModelNoOnboardingForm(request.POST, instance=self.object)
if form.is_valid():
form.save()
context = self.get_context_data(object=self.object, **kwargs)
return self.render_to_response(context)
[docs]
class DeviceCertificateLifecycleManagementSummaryView(AbstractCertificateLifecycleManagementSummaryView):
"""Certificate Lifecycle Management Summary View for devices."""
[docs]
page_name = DEVICES_PAGE_DEVICES_SUBCATEGORY
[docs]
def dispatch(self, request: HttpRequest, *args: Any, **kwargs: Any) -> HttpResponseBase:
"""Redirect OPC UA GDS Push devices to their specific view if GDS Push is the only protocol."""
device = self.get_object()
if (
device.device_type == DeviceModel.DeviceType.OPC_UA_GDS_PUSH
and device.onboarding_config
and not (
device.onboarding_config.has_pki_protocol(OnboardingPkiProtocol.CMP)
or device.onboarding_config.has_pki_protocol(OnboardingPkiProtocol.EST)
)
):
url = reverse('devices:opc_ua_gds_push_certificate_lifecycle_management', kwargs={'pk': device.pk})
return redirect(url)
return super().dispatch(request, *args, **kwargs)
[docs]
class OpcUaGdsCertificateLifecycleManagementSummaryView(AbstractCertificateLifecycleManagementSummaryView):
"""Certificate Lifecycle Management Summary View for OPC UA Devcies."""
[docs]
page_name = DEVICES_PAGE_OPC_UA_SUBCATEGORY
[docs]
class OpcUaGdsPushCertificateLifecycleManagementSummaryView(AbstractCertificateLifecycleManagementSummaryView):
"""Certificate Lifecycle Management Summary View for OPC UA GDS Push devices."""
[docs]
page_name = DEVICES_PAGE_DEVICES_SUBCATEGORY
[docs]
def get_context_data(self, **kwargs: Any) -> dict[str, Any]:
"""Add OPC UA GDS Push specific context.
Args:
**kwargs: Keyword arguments passed to super().get_context_data(**kwargs).
Returns:
The context data for the view.
"""
return super().get_context_data(**kwargs)
[docs]
def get_onboarding_initial(self) -> dict[str, Any]:
"""Gets the initial values for onboarding for GDS Push.
Returns:
Initial values for onboarding.
"""
if not self.object.onboarding_config:
err_msg = gettext_lazy('The device does not have onboarding configured.')
raise ValueError(err_msg)
return {
'common_name': self.object.common_name,
'serial_number': self.object.serial_number,
'domain': self.object.domain,
'ip_address': self.object.ip_address,
'opc_server_port': self.object.opc_server_port,
'onboarding_protocol': OnboardingProtocol.OPC_GDS_PUSH.label,
'onboarding_status': OnboardingStatus(self.object.onboarding_config.onboarding_status).label,
'pki_protocol_opc_gds_push': self.object.onboarding_config.has_pki_protocol(
OnboardingPkiProtocol.OPC_GDS_PUSH
),
}
[docs]
def post(self, request: HttpRequest, *_args: Any, **kwargs: Any) -> HttpResponse:
"""Handles the POST request used for device form submission.
Args:
request: The django request object.
_args: Positional arguments are discarded.
kwargs: Keyword arguments are passed to get_context_data.
Returns:
The HttpResponse.
"""
self.object = self.get_object()
if self.object.onboarding_config:
form = ClmDeviceModelOpcUaGdsPushOnboardingForm(request.POST, instance=self.object)
if form.is_valid():
form.save()
else:
# This shouldn't happen for GDS Push
pass
context = self.get_context_data(object=self.object, **kwargs)
return self.render_to_response(context)
# ------------------------------ Certificate Lifecycle Management - Credential Issuance -------------------------------
[docs]
class AbstractNoOnboardingIssueNewApplicationCredentialView(PageContextMixin, DetailView[DeviceModel]):
"""abc."""
[docs]
http_method_names = ('get',)
[docs]
context_object_name = 'device'
[docs]
template_name = 'devices/credentials/issue_credential.html'
[docs]
page_category = DEVICES_PAGE_CATEGORY
[docs]
def get_context_data(self, **kwargs: Any) -> dict[str, Any]:
"""Add the sections to the context.
Args:
**kwargs: Keyword arguments are passed to super().get_context_data(**kwargs).
Returns:
The context data for the view.
"""
context = super().get_context_data(**kwargs)
context['clm_url'] = f'{self.page_category}:{self.page_name}_certificate_lifecycle_management'
context['heading'] = 'Issue New Application Credential'
sections = []
if not self.object.no_onboarding_config:
err_msg = gettext_lazy('Device is configured for onboarding')
raise ValueError(err_msg)
sections.append(
{
'heading': gettext_lazy('CMP with OpenSSL (shared-secret)'),
'description': gettext_lazy(
'This option will guide you through all steps and commands that are '
'required to issue a new application certificate '
'using CMP with OpenSSL using a shared-secret (HMAC).'
),
'protocol': 'cmp-shared-secret',
'enabled': self.object.no_onboarding_config.has_pki_protocol(NoOnboardingPkiProtocol.CMP_SHARED_SECRET),
'url': f'{self.page_category}:{self.page_name}_no_onboarding_cmp_shared_secret_help',
}
)
sections.append(
{
'heading': gettext_lazy('EST with OpenSSL and curL (username & password)'),
'description': gettext_lazy(
'This option will guide you through all steps and commands that are '
'required to issue a new application certificate using EST using OpenSSL and curL.'
),
'protocol': 'est-username-password',
'enabled': self.object.no_onboarding_config.has_pki_protocol(
NoOnboardingPkiProtocol.EST_USERNAME_PASSWORD
),
'url': f'{self.page_category}:{self.page_name}_no_onboarding_est_username_password_help',
}
)
sections.append(
{
'heading': gettext_lazy('Manual Issuance'),
'description': gettext_lazy(
'This option will allow you to issue a new domain credential on the Trustpoint. '
'The domain credential can then be downloaded for manual injection into the device, '
'e.g., using a USB-stick.'
),
'protocol': 'manual',
'enabled': self.object.no_onboarding_config.has_pki_protocol(NoOnboardingPkiProtocol.MANUAL),
'url': f'{self.page_category}:{self.page_name}_no_onboarding_select_certificate_profile',
}
)
context['sections'] = sections
return context
def _get_redirect_url(self) -> HttpResponseRedirect:
return redirect(f'{self.page_category}:{self.page_name}_certificate_lifecycle_management', pk=self.object.pk)
[docs]
def get(self, request: HttpRequest, *_args: Any, **_kwargs: Any) -> HttpResponse:
"""Adds checks if the device is configured for no-onboarding and has a domain set.
Args:
request: The django request object.
*_args: Positional arguments are discarded.
**_kwargs: Keyword arguments are discarded.
Returns:
The HttpResponse Or HttpRedirect to the CLM page.
"""
self.object = self.get_object()
if not self.object.no_onboarding_config:
err_msg = gettext_lazy('This device is configured for onboarding.')
messages.warning(request, err_msg)
return self._get_redirect_url()
if not self.object.no_onboarding_config.get_pki_protocols():
err_msg = gettext_lazy(
'All PKI protocols for this device to request application certifciates are disabled.'
)
messages.warning(request, err_msg)
return self._get_redirect_url()
if not self.object.domain:
err_msg = gettext_lazy('No domain is configured for this device.')
messages.warning(request, err_msg)
return self._get_redirect_url()
context = self.get_context_data(object=self.object)
return self.render_to_response(context)
[docs]
class DeviceNoOnboardingIssueNewApplicationCredentialView(AbstractNoOnboardingIssueNewApplicationCredentialView):
"""abc."""
[docs]
page_name = DEVICES_PAGE_DEVICES_SUBCATEGORY
[docs]
class OpcUaGdsNoOnboardingIssueNewApplicationCredentialView(AbstractNoOnboardingIssueNewApplicationCredentialView):
"""abc."""
[docs]
page_name = DEVICES_PAGE_OPC_UA_SUBCATEGORY
[docs]
class AbstractSelectCertificateProfileNewApplicationCredentialView(PageContextMixin, DetailView[DeviceModel]):
"""abc."""
[docs]
http_method_names = ('get',)
[docs]
context_object_name = 'device'
[docs]
template_name = 'devices/credentials/profile_select.html'
[docs]
page_category = DEVICES_PAGE_CATEGORY
[docs]
def get_context_data(self, **kwargs: Any) -> dict[str, Any]:
"""Add the sections to the context.
Args:
**kwargs: Keyword arguments are passed to super().get_context_data(**kwargs).
Returns:
The context data for the view.
"""
context = super().get_context_data(**kwargs)
domain = self.object.domain
if not domain:
err_msg = gettext_lazy('No domain is configured for this device.')
raise ValueError(err_msg)
allowed_app_profiles = list(
domain.get_allowed_cert_profiles().exclude(certificate_profile__unique_name='domain_credential'))
profile_list = DomainAllowedCertificateProfileModel.get_list_of_display_names(allowed_app_profiles)
context['cert_profile_list'] = {}
for (profile_id, display_name, _unique_name) in profile_list:
context['cert_profile_list'][profile_id] = display_name
context['profile_issuance_url'] = \
f'{self.page_category}:{self.page_name}_certificate_lifecycle_management_issue_profile_credential'
return context
[docs]
class DeviceSelectCertificateProfileNewApplicationCredentialView(
AbstractSelectCertificateProfileNewApplicationCredentialView
):
"""abc."""
[docs]
page_name = DEVICES_PAGE_DEVICES_SUBCATEGORY
[docs]
class OpcUaGdsSelectCertificateProfileNewApplicationCredentialView(
AbstractSelectCertificateProfileNewApplicationCredentialView
):
"""abc."""
[docs]
page_name = DEVICES_PAGE_OPC_UA_SUBCATEGORY
[docs]
class AbstractOnboardingIssueNewApplicationCredentialView(PageContextMixin, DetailView[DeviceModel]):
"""abc."""
[docs]
http_method_names = ('get',)
[docs]
context_object_name = 'device'
[docs]
template_name = 'devices/credentials/issue_credential.html'
[docs]
page_category = DEVICES_PAGE_CATEGORY
[docs]
def get_context_data(self, **kwargs: Any) -> dict[str, Any]:
"""Add the sections to the context.
Args:
**kwargs: Keyword arguments are passed to super().get_context_data(**kwargs).
Returns:
The context data for the view.
"""
context = super().get_context_data(**kwargs)
context['clm_url'] = f'{self.page_category}:{self.page_name}_certificate_lifecycle_management'
context['heading'] = 'Issue New Application Credential'
sections = []
if not self.object.onboarding_config:
err_msg = gettext_lazy('Device is not configured for onboarding')
raise ValueError(err_msg)
sections.append(
{
'heading': gettext_lazy('CMP with Domain Credential'),
'description': gettext_lazy(
'This option will guide you through all steps and commands that are '
'required to issue a new application certificate using CMP '
'with OpenSSL using a shared-secret (HMAC).'
),
'protocol': 'cmp',
'enabled': self.object.onboarding_config.has_pki_protocol(OnboardingPkiProtocol.CMP),
'url': (
f'{self.page_category}:{self.page_name}_onboarding_clm_issue_application_credential_cmp_domain_credential'
),
}
)
sections.append(
{
'heading': gettext_lazy('EST with Domain Credential'),
'description': gettext_lazy(
'This option will guide you through all steps and commands that are '
'required to issue a new application certificate using EST using OpenSSL and curL.'
),
'protocol': 'est',
'enabled': self.object.onboarding_config.has_pki_protocol(OnboardingPkiProtocol.EST),
'url': (
f'{self.page_category}:{self.page_name}_onboarding_clm_issue_application_credential_est_domain_credential'
),
}
)
context['sections'] = sections
return context
[docs]
class DeviceOnboardingIssueNewApplicationCredentialView(AbstractOnboardingIssueNewApplicationCredentialView):
"""abc."""
[docs]
page_name = DEVICES_PAGE_DEVICES_SUBCATEGORY
[docs]
class OpcUaGdsOnboardingIssueNewApplicationCredentialView(AbstractOnboardingIssueNewApplicationCredentialView):
"""abc."""
[docs]
page_name = DEVICES_PAGE_OPC_UA_SUBCATEGORY
[docs]
class OpcUaGdsPushOnboardingIssueNewApplicationCredentialView(AbstractOnboardingIssueNewApplicationCredentialView):
"""View for issuing application credentials for OPC UA GDS Push devices - redirects directly to help page."""
[docs]
page_name = DEVICES_PAGE_DEVICES_SUBCATEGORY
[docs]
def get(self, request: HttpRequest, *_args: Any, **_kwargs: Any) -> HttpResponse:
"""Redirect directly to the GDS Push application certificate help page if GDS Push is the only protocol."""
self.object = self.get_object()
if not self.object.onboarding_config:
err_msg = gettext_lazy('Device is not configured for onboarding')
messages.warning(request, err_msg)
return redirect(
f'{self.page_category}:{self.page_name}_certificate_lifecycle_management',
pk=self.object.pk
)
if not self.object.onboarding_config.get_pki_protocols():
err_msg = gettext_lazy(
'All PKI protocols for this device to request application certificates are disabled.'
)
messages.warning(request, err_msg)
return redirect(
f'{self.page_category}:{self.page_name}_certificate_lifecycle_management',
pk=self.object.pk
)
if not self.object.domain:
err_msg = gettext_lazy('No domain is configured for this device.')
messages.warning(request, err_msg)
return redirect(
f'{self.page_category}:{self.page_name}_certificate_lifecycle_management',
pk=self.object.pk
)
# If CMP or EST are enabled, show the protocol selection page
if (
self.object.onboarding_config.has_pki_protocol(OnboardingPkiProtocol.CMP)
or self.object.onboarding_config.has_pki_protocol(OnboardingPkiProtocol.EST)
):
return super().get(request, *_args, **_kwargs)
# Otherwise, redirect directly to the GDS Push application certificate help page
return redirect(
f'{self.page_category}:{self.page_name}_onboarding_clm_issue_application_credential_opc_ua_gds_push_domain_credential',
pk=self.object.pk
)
[docs]
class AbstractIssueCredentialView[FormClass: forms.Form, IssuerClass: BaseTlsCredentialIssuer](
PageContextMixin, FormMixin[forms.Form], DetailView[DeviceModel]
):
"""Base view for all credential issuance views."""
[docs]
http_method_names = ('get', 'post')
[docs]
context_object_name = 'device'
[docs]
template_name = 'devices/credentials/issue_application_credential.html'
[docs]
issuer_class: type[IssuerClass]
[docs]
page_category = DEVICES_PAGE_CATEGORY
[docs]
def get_context_data(self, **kwargs: Any) -> dict[str, Any]:
"""Add the form to the context.
Args:
**kwargs: Keyword arguments are passed to super().get_context_data(**kwargs).
Returns:
The context data for the view.
"""
context = super().get_context_data(**kwargs)
if 'form' not in kwargs:
context['form'] = self.form_class(**self.get_form_kwargs())
context['clm_url'] = f'devices:{self.page_name}_certificate_lifecycle_management'
return context
@abc.abstractmethod
[docs]
def issue_credential(self, device: DeviceModel, form: forms.Form) -> IssuedCredentialModel:
"""Abstract method to issue a credential.
Args:
device: The device to be associated with the new credential.
form: The form instance containing the validated data.
Returns:
The IssuedCredentialModel object that was created and saved.
"""
[docs]
def post(self, request: HttpRequest, *_args: Any, **_kwargs: Any) -> HttpResponse:
"""Adds the object model to the instance and forwards to super().post().
Args:
request: The Django request object is only used implicitly through self.
*_args: Positional arguments are discarded.
**_kwargs: Keyword arguments are discarded.
Returns:
The HttpResponseBase object returned by super().post().
"""
self.object = self.get_object()
form = self.form_class(**self.get_form_kwargs())
if form.is_valid():
try:
credential = self.issue_credential(device=self.object, form=form)
del credential # result unused beyond this point
messages.success(
request, f'Successfully issued {self.friendly_name} for device {self.object.common_name}'
)
except Exception as e: # noqa: BLE001
messages.error(request, f'Failed to issue {self.friendly_name}: {e}')
return HttpResponseRedirect(
reverse_lazy(
f'devices:{self.page_name}_certificate_lifecycle_management', kwargs={'pk': self.get_object().id}
)
)
return self.render_to_response(self.get_context_data(form=form))
[docs]
class AbstractIssueDomainCredentialView(
AbstractIssueCredentialView[IssueDomainCredentialForm, LocalDomainCredentialIssuer]
):
"""Base view for issuing domain credentials."""
[docs]
form_class: type[BaseCredentialForm] = IssueDomainCredentialForm
[docs]
template_name = 'devices/credentials/issue_domain_credential.html'
[docs]
issuer_class = LocalDomainCredentialIssuer
[docs]
friendly_name = 'Domain Credential'
[docs]
def issue_credential(self, device: DeviceModel, _form: forms.Form) -> IssuedCredentialModel:
"""Issue a domain credential for the device.
Args:
device: The device to issue the credential for.
_form: The form instance containing the validated data.
Returns:
The issued credential model.
"""
if device.domain is None:
err_msg = gettext_lazy('Device has no domain configured.')
raise Http404(err_msg)
issuer = self.issuer_class(device=device, domain=device.domain)
return issuer.issue_domain_credential()
[docs]
class DeviceIssueDomainCredentialView(AbstractIssueDomainCredentialView):
"""View for issuing domain credentials for devices."""
[docs]
page_name = DEVICES_PAGE_DEVICES_SUBCATEGORY
[docs]
class OpcUaGdsIssueDomainCredentialView(AbstractIssueDomainCredentialView):
"""View for issuing domain credentials for OPC-UA GDS devices."""
[docs]
page_name = DEVICES_PAGE_OPC_UA_SUBCATEGORY
[docs]
class OpcUaGdsPushIssueDomainCredentialView(AbstractIssueDomainCredentialView):
"""View for issuing domain credentials for OPC-UA GDS Push devices."""
[docs]
form_class = IssueOpcUaGdsPushDomainCredentialForm
[docs]
page_name = DEVICES_PAGE_DEVICES_SUBCATEGORY
[docs]
def issue_credential(self, device: DeviceModel, form: forms.Form) -> IssuedCredentialModel:
"""Issues a domain credential for the device with OPC UA specific extensions.
Args:
device: The device to be associated with the new credential.
form: The form instance containing the validated data.
Returns:
The IssuedCredentialModel object that was created and saved.
"""
if not device.domain:
raise Http404(DeviceWithoutDomainErrorMsg)
issuer = self.issuer_class(device=device, domain=device.domain)
application_uri = cast('str', form.cleaned_data.get('application_uri'))
opc_ua_extensions = [
(
x509.KeyUsage(
digital_signature=True,
content_commitment=True,
key_encipherment=True, # Required for OPC UA SignAndEncrypt security mode
data_encipherment=True, # Additional encryption capability
key_agreement=False,
key_cert_sign=False,
crl_sign=False,
encipher_only=False,
decipher_only=False,
),
True,
),
(x509.ExtendedKeyUsage([x509.oid.ExtendedKeyUsageOID.CLIENT_AUTH]), False),
]
if application_uri:
opc_ua_extensions.append(
(x509.SubjectAlternativeName([x509.UniformResourceIdentifier(application_uri)]), False)
)
return issuer.issue_domain_credential(
application_uri=None,
extra_extensions=opc_ua_extensions
)
[docs]
def post(self, request: HttpRequest, *_args: Any, **_kwargs: Any) -> HttpResponse:
"""Handle the POST request with custom redirect to truststore association."""
self.object = self.get_object()
form = self.form_class(**self.get_form_kwargs())
if form.is_valid():
self.issue_credential(device=self.object, form=form)
messages.success(
request, f'Successfully issued {self.friendly_name} for device {self.object.common_name}'
)
return HttpResponseRedirect(
reverse_lazy(
f'devices:{self.page_name}_truststore_association', kwargs={'pk': self.get_object().id}
)
)
return self.render_to_response(self.get_context_data(form=form))
[docs]
class OpcUaGdsPushDiscoverServerView(PageContextMixin, DetailView[DeviceModel]):
"""View to discover OPC UA server information without authentication."""
[docs]
http_method_names = ('post',)
[docs]
context_object_name = 'device'
[docs]
page_name = DEVICES_PAGE_DEVICES_SUBCATEGORY
[docs]
def post(self, request: HttpRequest, *_args: Any, **_kwargs: Any) -> HttpResponse:
"""Handle the POST request to discover server information."""
self.object = self.get_object()
try:
service = GdsPushService(device=self.object, insecure=True)
success, message, server_info = asyncio.run(service.discover_server())
if success and server_info:
messages.success(request, message)
request.session['opc_ua_server_info'] = server_info
if server_info.get('server_certificate_available'):
messages.info(
request,
'Server certificate found. You can update the truststore with this certificate.'
)
else:
messages.error(request, f'Failed to discover server: {message}')
except Exception as e: # noqa: BLE001
messages.error(request, f'Unexpected error during discovery: {e}')
return HttpResponseRedirect(
reverse_lazy(
f'devices:{self.page_name}_onboarding_truststore_associated_help',
kwargs={'pk': self.object.pk}
)
)
[docs]
class OpcUaGdsPushTruststoreAssociationView(
LoggerMixin, PageContextMixin, FormView[OpcUaGdsPushTruststoreAssociationForm]
):
"""View for associating a truststore with an OPC UA GDS Push device's onboarding configuration."""
[docs]
template_name = 'devices/truststore_association.html'
[docs]
page_name = DEVICES_PAGE_DEVICES_SUBCATEGORY
[docs]
def get_device(self) -> DeviceModel:
"""Get the device from the URL parameters."""
pk = self.kwargs.get('pk')
try:
return DeviceModel.objects.get(pk=pk)
except DeviceModel.DoesNotExist as e:
exc_msg = f'Device with pk {pk} does not exist.'
raise Http404(exc_msg) from e
[docs]
def get_context_data(self, **kwargs: Any) -> dict[str, Any]:
"""Add the device and import form to the context."""
context = super().get_context_data(**kwargs)
context['device'] = self.get_device()
context['import_form'] = TruststoreAddForm()
return context
[docs]
def post(self, request: HttpRequest, *args: Any, **kwargs: Any) -> HttpResponse:
"""Handle both association and import form submissions."""
if 'trust_store_file' in request.FILES:
return self._handle_import(request)
return super().post(request, *args, **kwargs)
def _handle_import(self, request: HttpRequest) -> HttpResponse:
"""Handle truststore import from modal."""
import_form = TruststoreAddForm(request.POST, request.FILES)
if import_form.is_valid():
truststore = import_form.cleaned_data['truststore']
n_certificates = truststore.number_of_certificates
msg_str = ngettext(
'Successfully created the Truststore %(name)s with %(count)i certificate.',
'Successfully created the Truststore %(name)s with %(count)i certificates.',
n_certificates,
) % {
'name': truststore.unique_name,
'count': n_certificates,
}
messages.success(request, msg_str)
return HttpResponseRedirect(
reverse('devices:devices_truststore_association', kwargs={'pk': self.get_device().pk}) +
f'?truststore_id={truststore.id}'
)
context = self.get_context_data()
context['import_form'] = import_form
return self.render_to_response(context)
[docs]
class AbstractIssueProfileCredentialView(
AbstractIssueCredentialView[CertificateIssuanceForm, BaseTlsCredentialIssuer]
):
"""View to issue a new certificate profile credential."""
[docs]
issuer_class = BaseTlsCredentialIssuer
[docs]
friendly_name = 'Certificate Profile Credential'
[docs]
template_name = 'pki/cert_profiles/issuance.html'
[docs]
def dispatch(self, request: HttpRequest, *args: Any, **kwargs: Any) -> HttpResponse:
"""Dispatch the request, ensuring the profile exists."""
self.profile = get_object_or_404(CertificateProfileModel, pk=kwargs['profile_id'])
return cast('HttpResponse', super().dispatch(request, *args, **kwargs))
[docs]
def get_context_data(self, **kwargs: Any) -> dict[str, Any]:
"""Add additional context data."""
context = super().get_context_data(**kwargs)
context['profile'] = self.profile
context['profile_dict'] = self.get_form_kwargs()['profile']
return context
[docs]
def issue_credential(self, device: DeviceModel, form: forms.Form
) -> IssuedCredentialModel:
"""Issues a credential based on the selected certificate profile.
Args:
device: The device to be associated with the new credential.
form: The form instance containing the validated data.
Returns:
The IssuedCredentialModel object that was created and saved.
"""
if not isinstance(form, CertificateIssuanceForm):
err_msg = 'Invalid form type. Expected CertificateIssuanceForm.'
raise TypeError(err_msg)
cert_builder = form.get_certificate_builder()
if not device.domain:
raise Http404(DeviceWithoutDomainErrorMsg)
ctx = ManualCredentialRequestContext(
device=device,
domain=device.domain,
cert_profile_str=self.profile.unique_name,
cert_requested=cert_builder,
)
ManualAuthorization().authorize(ctx)
CredentialIssueProcessor().process_operation(ctx)
if not ctx.issued_credential:
err_msg = 'Issued credential not in context.'
raise ValueError(err_msg)
return ctx.issued_credential
[docs]
class DeviceIssueProfileCredentialView(AbstractIssueProfileCredentialView):
"""Issue a new certificate profile credential within the devices section."""
[docs]
page_name = DEVICES_PAGE_DEVICES_SUBCATEGORY
[docs]
class OpcUaGdsIssueProfileCredentialView(AbstractIssueProfileCredentialView):
"""Issue a new certificate profile credential within the devices section."""
[docs]
page_name = DEVICES_PAGE_OPC_UA_SUBCATEGORY
# -------------------------------- Certificate Lifecycle Management - Token Auth Mixin --------------------------------
[docs]
class DownloadTokenRequiredAuthenticationMixin:
"""Mixin which checks the token included in the URL for browser download views."""
[docs]
credential_download: RemoteDeviceCredentialDownloadModel
[docs]
def dispatch(self, request: HttpRequest, *args: Any, **kwargs: Any) -> HttpResponseBase:
"""Checks the validity of the token included in the URL for browser download views and redirects if invalid.
Args:
request: The django request object.
*args: Positional arguments passed to super().dispatch().
**kwargs: Keyword arguments passed to super().dispatch().
Returns:
A Django HttpResponseBase object.
"""
super_dispatch = getattr(super(), 'dispatch', None)
if not callable(super_dispatch):
err_msg = 'Internal server error. Failed to get super().dispatch().'
raise Http404(err_msg)
token = request.GET.get('token')
try:
self.credential_download = RemoteDeviceCredentialDownloadModel.objects.get(
issued_credential_model=kwargs.get('pk')
)
except RemoteDeviceCredentialDownloadModel.DoesNotExist:
messages.warning(request, 'Invalid download token.')
return redirect('devices:browser_login')
if not token or not self.credential_download.check_token(token):
messages.warning(request, 'Invalid download token.')
return redirect('devices:browser_login')
return cast('HttpResponseBase', super_dispatch(request, *args, **kwargs))
# ----------------------------------- Certificate Lifecycle Management - Downloads ------------------------------------
[docs]
class AbstractDownloadPageDispatcherView(PageContextMixin, RedirectView):
"""Redirects depending on the type of credential, that is if a private key is available or not."""
[docs]
http_method_names = ('get',)
[docs]
model: type[IssuedCredentialModel] = IssuedCredentialModel
[docs]
page_category = DEVICES_PAGE_CATEGORY
[docs]
def get_redirect_url(self, *_args: Any, **kwargs: Any) -> str:
"""Gets the redirection URL depending on the type credential, that is if a private key is available or not.
Args:
*_args: Positional arguments are discarded.
**kwargs: The pk parameter is retrieved and expected to be there.
Returns:
The redirect URL.
"""
pk = kwargs.get('pk')
# This can only happen if the path for the URL defined in urls.py does not contain <int:pk>.
# This would mean we, the dev team, introduced a bug.
if pk is None or not isinstance(pk, int):
err_msg = 'An unexpected error occurred. Please see logs for more information.'
raise Http404(err_msg)
issued_credential = IssuedCredentialModel.objects.filter(pk=pk).first()
if issued_credential is None:
messages.error(
self.request, 'No credential found for the given primary key. See logs for more information.'
)
return reverse(f'devices:{self.page_name}_certificate_lifecycle_management', kwargs={'pk': pk})
if issued_credential.credential.private_key:
return reverse(f'devices:{self.page_name}_credential-download', kwargs={'pk': pk})
return reverse(f'devices:{self.page_name}_certificate-download', kwargs={'pk': pk})
[docs]
class DeviceDownloadPageDispatcherView(AbstractDownloadPageDispatcherView):
"""Download dispatcher view for the device pages."""
[docs]
page_name = DEVICES_PAGE_DEVICES_SUBCATEGORY
[docs]
class OpcUaGdsDownloadPageDispatcherView(AbstractDownloadPageDispatcherView):
"""Download dispatcher view for the OPC UA GDS pages."""
[docs]
page_name = DEVICES_PAGE_OPC_UA_SUBCATEGORY
[docs]
class OpcUaGdsPushDownloadPageDispatcherView(AbstractDownloadPageDispatcherView):
"""Download dispatcher view for the OPC UA GDS Push pages."""
[docs]
page_name = DEVICES_PAGE_DEVICES_SUBCATEGORY
# --------------------------------------------- Certificate Download Help ----------------------------------------------
[docs]
class AbstractCertificateDownloadView(PageContextMixin, DetailView[IssuedCredentialModel]):
"""View for downloading certificates."""
[docs]
http_method_names = ('get',)
[docs]
model: type[IssuedCredentialModel] = IssuedCredentialModel
[docs]
template_name = 'devices/credentials/certificate_download.html'
[docs]
context_object_name = 'issued_credential'
[docs]
page_category = DEVICES_PAGE_CATEGORY
[docs]
def get_context_data(self, **kwargs: Any) -> dict[str, Any]:
"""Add the clm_url to the context.
Args:
**kwargs: Keyword arguments are passed to super().get_context_data(**kwargs).
Returns:
The context data for the view.
"""
context = super().get_context_data(**kwargs)
context['clm_url'] = f'{self.page_category}:{self.page_name}_certificate_lifecycle_management'
return context
[docs]
class DeviceCertificateDownloadView(AbstractCertificateDownloadView):
"""Certificate download view for the device pages."""
[docs]
page_name = DEVICES_PAGE_DEVICES_SUBCATEGORY
[docs]
class OpcUaGdsCertificateDownloadView(AbstractCertificateDownloadView):
"""Certificate download view for the OPC UA GDS pages."""
[docs]
page_name = DEVICES_PAGE_OPC_UA_SUBCATEGORY
[docs]
class OpcUaGdsPushCertificateDownloadView(AbstractCertificateDownloadView):
"""Certificate download view for the OPC UA GDS Push pages."""
[docs]
page_name = DEVICES_PAGE_DEVICES_SUBCATEGORY
# -------------------------------------------- OPC UA GDS Push Actions -------------------------------------------------
[docs]
class OpcUaGdsPushUpdateTrustlistView(PageContextMixin, DetailView[DeviceModel]):
"""View to update the trustlist on an OPC UA GDS Push device."""
[docs]
http_method_names = ('post',)
[docs]
page_name = DEVICES_PAGE_DEVICES_SUBCATEGORY
[docs]
page_category = DEVICES_PAGE_CATEGORY
[docs]
def post(self, request: HttpRequest, *_args: Any, **_kwargs: Any) -> HttpResponse:
"""Handle the POST request to update the trustlist.
Args:
request: The Django request object.
_args: Positional arguments are discarded.
kwargs: Keyword arguments are passed to get_object().
Returns:
HttpResponse redirecting back to the help page.
"""
self.object = self.get_object()
try:
service = GdsPushService(device=self.object)
success, message = asyncio.run(service.update_trustlist())
if success:
messages.success(request, f'Trustlist updated successfully: {message}')
else:
messages.error(request, f'Failed to update trustlist: {message}')
except GdsPushError as e:
messages.error(request, f'GDS Push error: {e}')
except Exception as e: # noqa: BLE001
messages.error(request, f'Unexpected error: {e}')
return self._redirect_to_help_page()
def _redirect_to_help_page(self) -> HttpResponseRedirect:
"""Redirect back to the help page.
Returns:
HttpResponseRedirect to the help page.
"""
return HttpResponseRedirect(
reverse_lazy(
f'{self.page_category}:{self.page_name}_onboarding_truststore_associated_help',
kwargs={'pk': self.object.pk}
)
)
[docs]
class OpcUaGdsPushUpdateServerCertificateView(PageContextMixin, DetailView[DeviceModel]):
"""View to update the server certificate on an OPC UA GDS Push device."""
[docs]
http_method_names = ('post',)
[docs]
page_name = DEVICES_PAGE_DEVICES_SUBCATEGORY
[docs]
page_category = DEVICES_PAGE_CATEGORY
[docs]
def post(self, request: HttpRequest, *_args: Any, **_kwargs: Any) -> HttpResponse:
"""Handle the POST request to update the server certificate.
Args:
request: The Django request object.
_args: Positional arguments are discarded.
kwargs: Keyword arguments are passed to get_object().
Returns:
HttpResponse redirecting back to the help page.
"""
self.object = self.get_object()
try:
service = GdsPushService(device=self.object)
success, message, certificate_bytes = asyncio.run(service.update_server_certificate())
if success and certificate_bytes:
messages.success(request, f'Server certificate updated successfully: {message}')
if self.object.onboarding_config:
self.object.onboarding_config.onboarding_status = OnboardingStatus.ONBOARDED
self.object.onboarding_config.save()
else:
messages.error(request, f'Failed to update server certificate: {message}')
except GdsPushError as e:
messages.error(request, f'GDS Push error: {e}')
except Exception as e: # noqa: BLE001
messages.error(request, f'Unexpected error: {e}')
return self._redirect_to_help_page()
def _update_truststore_with_certificate(self, certificate_bytes: bytes) -> None:
"""Update or create truststore with the new server certificate.
This method adds the new server certificate to the truststore WITHOUT removing
the old one immediately. This ensures that if something goes wrong, the old
certificate is still available for recovery.
Args:
certificate_bytes: DER-encoded certificate bytes
"""
if self.object.onboarding_config is None:
msg = 'Onboarding config must exist for GDS Push devices'
raise ValueError(msg)
try:
cert = x509.load_der_x509_certificate(certificate_bytes, default_backend())
truststore = self.object.onboarding_config.opc_trust_store
if not truststore:
truststore = TruststoreModel.objects.create(
unique_name=f'opc_server_{self.object.common_name}',
intended_usage=TruststoreModel.IntendedUsage.OPC_UA_GDS_PUSH
)
self.object.onboarding_config.opc_trust_store = truststore
self.object.onboarding_config.save()
existing_certs = list(truststore.certificates.all())
cert_serial = cert.serial_number
cert_fingerprint = cert.fingerprint(hashes.SHA256())
cert_exists = False
for existing_cert_model in existing_certs:
existing_cert = existing_cert_model.get_certificate_serializer().as_crypto()
if (existing_cert.serial_number == cert_serial and
existing_cert.fingerprint(hashes.SHA256()) == cert_fingerprint):
cert_exists = True
break
if not cert_exists:
for existing_cert_model in truststore.certificates.all():
if not existing_cert_model.is_ca:
RevokedCertificateModel.objects.create(
certificate=existing_cert_model,
revocation_reason=RevokedCertificateModel.ReasonCode.CESSATION
)
truststore.truststoreordermodel_set.all().delete()
cert_model = CertificateModel.save_certificate(cert)
TruststoreOrderModel.objects.create(
trust_store=truststore,
certificate=cert_model,
order=0
)
messages.info(
self._get_request(),
f'Updated truststore "{truststore.unique_name}" with new server certificate '
f'(total: {truststore.certificates.count()} cert(s))'
)
else:
messages.info(
self._get_request(),
f'Server certificate already exists in truststore "{truststore.unique_name}"'
)
except Exception as e: # noqa: BLE001
messages.warning(
self._get_request(),
f'Server certificate updated but failed to update truststore: {e}'
)
def _get_request(self) -> HttpRequest:
"""Get the current request from view context.
Returns:
The current HttpRequest
"""
return self.request
def _redirect_to_help_page(self) -> HttpResponseRedirect:
"""Redirect back to the help page.
Returns:
HttpResponseRedirect to the help page.
"""
referer = self.request.META.get('HTTP_REFERER', '')
if referer and 'issue-application-credential' in referer:
return HttpResponseRedirect(
reverse_lazy(
f'{self.page_category}:{self.page_name}_onboarding_clm_issue_application_credential_opc_ua_gds_push_domain_credential',
kwargs={'pk': self.object.pk}
)
)
return HttpResponseRedirect(
reverse_lazy(
f'{self.page_category}:{self.page_name}_onboarding_truststore_associated_help',
kwargs={'pk': self.object.pk}
)
)
[docs]
class OpcUaGdsPushCertRenewalSettingsView(PageContextMixin, DetailView[DeviceModel]):
"""View to save the periodic server certificate and trustlist renewal settings for an OPC UA GDS Push device."""
[docs]
http_method_names = ('post',)
[docs]
page_name = DEVICES_PAGE_DEVICES_SUBCATEGORY
[docs]
page_category = DEVICES_PAGE_CATEGORY
[docs]
def post(self, request: HttpRequest, *_args: Any, **_kwargs: Any) -> HttpResponse:
"""Handle the POST request to update the renewal settings.
Args:
request: The Django request object.
_args: Positional arguments are discarded.
_kwargs: Keyword arguments are discarded.
Returns:
HttpResponse redirecting back to the help page.
"""
self.object = self.get_object()
enable = request.POST.get('opc_gds_push_enable_periodic_update') == 'on'
interval_raw = request.POST.get('opc_gds_push_renewal_interval', '')
try:
interval = int(interval_raw)
if interval < 1:
err_msg = 'Interval must be at least 1.'
raise ValueError(err_msg) # noqa: TRY301
except (ValueError, TypeError):
messages.error(
request,
gettext_lazy('Invalid renewal interval. Please enter a positive integer (hours).')
)
return self._redirect_to_help_page()
self.object.opc_gds_push_enable_periodic_update = enable
self.object.opc_gds_push_renewal_interval = interval
self.object.save(
update_fields=['opc_gds_push_enable_periodic_update', 'opc_gds_push_renewal_interval']
)
if enable:
messages.success(
request,
gettext_lazy(
f'Periodic server certificate and trustlist renewal enabled (every {interval} hour(s)).'
)
)
else:
messages.success(
request,
gettext_lazy('Periodic server certificate and trustlist renewal disabled.')
)
return self._redirect_to_help_page()
def _redirect_to_help_page(self) -> HttpResponseRedirect:
"""Redirect back to the application certificate help page.
Returns:
HttpResponseRedirect to the help page.
"""
return HttpResponseRedirect(
reverse_lazy(
f'{self.page_category}:{self.page_name}_onboarding_clm_issue_application_credential_opc_ua_gds_push_domain_credential',
kwargs={'pk': self.object.pk}
)
)
[docs]
class TrustBundleDownloadView(PageContextMixin, DetailView[CaModel]):
"""View to download the trust bundle (CA certificates and CRLs) for a given Issuing CA."""
[docs]
http_method_names = ('get',)
def _collect_ca_chain_data(self, ca_chain: list[CaModel]) -> dict[str, bytes]:
"""Collect certificate and CRL data from CA chain.
Args:
ca_chain: List of CA models in the chain.
Returns:
Dictionary mapping filenames to file data (bytes).
"""
data_to_archive: dict[str, bytes] = {}
for ca in ca_chain:
try:
if not ca.ca_certificate_model:
continue
cert_der = ca.ca_certificate_model.get_certificate_serializer().as_der()
if cert_der is None:
continue
safe_name = ''.join(c if c.isalnum() or c in '._-' else '_' for c in ca.unique_name)
cert_filename = f'{safe_name}.der'
data_to_archive[cert_filename] = cert_der
if ca.crl_pem:
try:
crl_crypto = x509.load_pem_x509_crl(ca.crl_pem.encode())
crl_der = crl_crypto.public_bytes(serialization.Encoding.DER)
crl_filename = f'{safe_name}.crl'
data_to_archive[crl_filename] = crl_der
except (ValueError, TypeError):
pass
except (ValueError, TypeError, AttributeError):
continue
return data_to_archive
def _create_trust_bundle_zip(self, data_to_archive: dict[str, bytes]) -> bytes:
"""Create a ZIP file containing trust bundle data.
Args:
data_to_archive: Dictionary mapping filenames to file data.
Returns:
The ZIP file as bytes.
Raises:
Http404: If no data or ZIP creation fails.
"""
if not data_to_archive:
msg = 'No certificates found in truststore.'
raise Http404(msg)
zip_buffer = io.BytesIO()
with zipfile.ZipFile(zip_buffer, 'w', zipfile.ZIP_DEFLATED) as zip_file:
for filename, data in data_to_archive.items():
zip_file.writestr(filename, data)
zip_data = zip_buffer.getvalue()
if not zip_data or len(zip_data) == 0:
msg = f'Failed to create ZIP file. Found {len(data_to_archive)} files to archive.'
raise Http404(msg)
return zip_data
[docs]
def get(self, request: HttpRequest, *args: Any, **kwargs: Any) -> HttpResponse: # noqa: ARG002
"""Handle the GET request to download the trust bundle.
Args:
request: The Django request object.
*args: Additional positional arguments.
**kwargs: Additional keyword arguments.
Returns:
FileResponse with the trust bundle zip file.
"""
self.object = self.get_object()
issuing_ca = self.object
try:
ca_chain = issuing_ca.get_ca_chain_from_truststore()
except ValueError as e:
msg = f'Invalid truststore configuration: {e}'
raise Http404(msg) from e
data_to_archive = self._collect_ca_chain_data(ca_chain)
zip_data = self._create_trust_bundle_zip(data_to_archive)
response = FileResponse(
io.BytesIO(zip_data),
content_type='application/zip',
as_attachment=True,
filename=f'trustpoint-{self.object.unique_name}-trust-bundle.zip',
)
return cast('HttpResponse', response)
# ---------------------------------------------- Credential Download Help ----------------------------------------------
[docs]
class AbstractDeviceBaseCredentialDownloadView(PageContextMixin, DetailView[IssuedCredentialModel]):
"""View to download a password protected application credential in the desired format.
Inherited by the domain and application credential download views. It is not intended for direct use.
"""
[docs]
http_method_names = ('get', 'post')
[docs]
model = IssuedCredentialModel
[docs]
template_name = 'devices/credentials/credential_download.html'
[docs]
context_object_name = 'credential'
[docs]
is_browser_download = False
[docs]
page_category = DEVICES_PAGE_CATEGORY
[docs]
def get_context_data(self, **kwargs: Any) -> dict[str, Any]:
"""Adds information about the credential to the context.
Args:
**kwargs: Keyword arguments are passed to super().get_context_data(**kwargs).
Returns:
The context data for the view.
"""
context = super().get_context_data(**kwargs)
issued_credential = self.object
credential = issued_credential.credential
if credential.credential_type != CredentialModel.CredentialTypeChoice.ISSUED_CREDENTIAL: # sanity check
err_msg = 'Credential is not an issued credential.'
raise Http404(err_msg)
cert_profile_name = issued_credential.issued_using_cert_profile
domain_credential_value = IssuedCredentialModel.IssuedCredentialType.DOMAIN_CREDENTIAL.value
application_credential_value = IssuedCredentialModel.IssuedCredentialType.APPLICATION_CREDENTIAL.value
if issued_credential.issued_credential_type == domain_credential_value:
context['credential_type'] = cert_profile_name
elif issued_credential.issued_credential_type == application_credential_value:
context['credential_type'] = cert_profile_name + ' Credential'
else:
err_msg = 'Unknown IssuedCredentialType'
raise Http404(err_msg)
context['FileFormat'] = CredentialFileFormat.__members__
context['is_browser_dl'] = self.is_browser_download
context['show_browser_dl'] = not self.is_browser_download
context['issued_credential'] = issued_credential
context['suggested_password'] = CredentialDownloadForm.get_suggested_password()
if 'form' not in kwargs:
context['form'] = self.form_class()
else:
context['form'] = kwargs['form']
context['browser_otp_url'] = f'devices:{self.page_name}_browser_otp_view'
context['clm_url'] = f'devices:{self.page_name}_certificate_lifecycle_management'
return context
[docs]
def post(self, _request: HttpRequest, *_args: Any, **_kwargs: Any) -> HttpResponse:
"""Processing the valid form data.
This will use the contained form data to start the download process of the desired file.
Args:
_request: The django request object.
*_args: Positional arguments are discarded.
**_kwargs: Keyword arguments are discarded.
Returns:
If successful, this will start the file download. Otherwise, a Http404 will be raised and displayed.
"""
self.object = self.get_object()
form = self.form_class(self.request.POST)
if form.is_valid():
password = form.cleaned_data['password'].encode()
try:
file_format = CredentialFileFormat(self.request.POST.get('file_format'))
except ValueError as exception:
err_msg = gettext_lazy('Unknown file format.')
raise Http404(err_msg) from exception
credential_model = self.object.credential
credential_serializer = credential_model.get_credential_serializer()
private_key_serializer = credential_serializer.get_private_key_serializer()
certificate_serializer = credential_serializer.get_certificate_serializer()
cert_collection_serializer = credential_serializer.get_additional_certificates_serializer()
if not private_key_serializer or not certificate_serializer or not cert_collection_serializer:
raise Http404
cert_profile_name = self.object.issued_using_cert_profile
credential_type_name = cert_profile_name.replace(' ', '-').lower().replace('-credential', '')
if file_format == CredentialFileFormat.PKCS12:
file_stream_data = io.BytesIO(credential_serializer.as_pkcs12(password=password))
elif file_format == CredentialFileFormat.PEM_ZIP:
file_data = Archiver.archive_zip(
data_to_archive={
'private_key.pem': private_key_serializer.as_pkcs8_pem(password=password),
'certificate.pem': certificate_serializer.as_pem(),
'certificate_chain.pem': cert_collection_serializer.as_pem(),
}
)
file_stream_data = io.BytesIO(file_data)
elif file_format == CredentialFileFormat.PEM_TAR_GZ:
file_data = Archiver.archive_tar_gz(
data_to_archive={
'private_key.pem': private_key_serializer.as_pkcs8_pem(password=password),
'certificate.pem': certificate_serializer.as_pem(),
'certificate_chain.pem': cert_collection_serializer.as_pem(),
}
)
file_stream_data = io.BytesIO(file_data)
else:
err_msg = gettext_lazy('Unknown file format.')
raise Http404(err_msg)
response = FileResponse(
file_stream_data,
content_type=file_format.mime_type,
as_attachment=True,
filename=f'trustpoint-{credential_type_name}-credential{file_format.file_extension}',
)
return cast('HttpResponse', response)
return self.render_to_response(self.get_context_data(form=form))
[docs]
class DeviceManualCredentialDownloadView(AbstractDeviceBaseCredentialDownloadView):
"""View to download a password protected domain or application credential in the desired format."""
[docs]
page_name = DEVICES_PAGE_DEVICES_SUBCATEGORY
@method_decorator(login_not_required, name='dispatch')
[docs]
class DeviceBrowserCredentialDownloadView(
DownloadTokenRequiredAuthenticationMixin, AbstractDeviceBaseCredentialDownloadView
):
"""View to download a password protected domain or app credential in the desired format from a remote client."""
[docs]
is_browser_download = True
[docs]
class AbstractBrowserOnboardingOTPView(PageContextMixin, DetailView[IssuedCredentialModel]):
"""View to display the OTP for remote credential download (aka. browser onboarding)."""
[docs]
http_method_names = ('get',)
[docs]
model = IssuedCredentialModel
[docs]
template_name = 'devices/credentials/onboarding/browser/otp_view.html'
[docs]
redirection_view = 'devices:devices'
[docs]
context_object_name = 'credential'
[docs]
page_category = DEVICES_PAGE_CATEGORY
[docs]
def get_context_data(self, **kwargs: Any) -> dict[str, Any]:
"""Adds information about the credential and otp for the browser download process.
Args:
**kwargs: Keyword arguments passed to super().get_context_data.
Returns:
The context to render the page.
"""
credential = self.get_object()
device = credential.device
if device is None:
raise Http404
context = super().get_context_data(**kwargs)
try:
cdm = RemoteDeviceCredentialDownloadModel.objects.get(issued_credential_model=credential, device=device)
cdm.delete()
except RemoteDeviceCredentialDownloadModel.DoesNotExist:
pass
cdm = RemoteDeviceCredentialDownloadModel(issued_credential_model=credential, device=device)
cdm.save()
context.update(
{
'device_name': device.common_name,
'device_id': device.id,
'credential_id': credential.id,
'otp': cdm.get_otp_display(),
'download_url': self.request.build_absolute_uri(reverse('devices:browser_login')),
}
)
context['cred_download_url'] = f'devices:{self.page_name}_credential-download'
context['browser_cancel'] = f'devices:{self.page_name}_browser_cancel'
return context
[docs]
class DeviceBrowserOnboardingOTPView(AbstractBrowserOnboardingOTPView):
"""The browser onboarding OTP view for the devices section."""
[docs]
page_name = DEVICES_PAGE_DEVICES_SUBCATEGORY
[docs]
class OpcUaGdsBrowserOnboardingOTPView(AbstractBrowserOnboardingOTPView):
"""The browser onboarding OTP view for the OPC UA GDS section."""
[docs]
page_name = DEVICES_PAGE_OPC_UA_SUBCATEGORY
[docs]
class AbstractBrowserOnboardingCancelView(PageContextMixin, DetailView[IssuedCredentialModel]):
"""View to cancel the browser onboarding process and delete the associated RemoteDeviceCredentialDownloadModel."""
[docs]
http_method_names = ('get',)
[docs]
model = IssuedCredentialModel
[docs]
context_object_name = 'credential'
[docs]
page_category = DEVICES_PAGE_CATEGORY
[docs]
def get(self, request: HttpRequest, *_args: Any, **_kwargs: Any) -> HttpResponse:
"""Cancels the browser onboarding process and deletes the associated RemoteDeviceCredentialDownloadModel.
Args:
request: The Django request object.
*_args: Positional arguments are discarded.
**_kwargs: Keyword arguments are discarded.
Returns:
The HttpResponseBase object with the desired redirection URL.
"""
self.object = self.get_object()
try:
cdm = RemoteDeviceCredentialDownloadModel.objects.get(
issued_credential_model=self.object, device=self.object.device
)
cdm.delete()
messages.info(request, 'The browser onboarding process was canceled.')
except RemoteDeviceCredentialDownloadModel.DoesNotExist:
pass
return HttpResponseRedirect(
reverse_lazy(f'devices:{self.page_name}_credential-download', kwargs={'pk': self.object.id})
)
[docs]
class DeviceBrowserOnboardingCancelView(AbstractBrowserOnboardingCancelView):
"""Cancels the browser onboarding for the devices section."""
[docs]
page_name = DEVICES_PAGE_DEVICES_SUBCATEGORY
[docs]
class OpcUaGdsBrowserOnboardingCancelView(AbstractBrowserOnboardingCancelView):
"""Cancels the browser onboarding for the OPC UA GDS section."""
[docs]
page_name = DEVICES_PAGE_OPC_UA_SUBCATEGORY
@method_decorator(login_not_required, name='dispatch')
[docs]
class DeviceOnboardingBrowserLoginView(FormView[BrowserLoginForm]):
"""View to handle certificate download requests."""
[docs]
http_method_names = ('get', 'post')
[docs]
template_name = 'devices/credentials/onboarding/browser/login.html'
[docs]
cleaned_data: dict[str, Any]
[docs]
def get_success_url(self) -> str:
"""Gets the success url to redirect to after successful processing of the POST data following a form submit.
Returns:
The success url to redirect to after successful processing of the POST data following a form submit.
"""
credential_id = cast('int', self.cleaned_data.get('credential_id'))
credential_download = cast('RemoteDeviceCredentialDownloadModel', self.cleaned_data.get('credential_download'))
token: str = credential_download.download_token
return (
f'{reverse_lazy("devices:browser_domain_credential_download", kwargs={"pk": credential_id})}?token={token}'
)
# ---------------------------------------- Revocation Views ----------------------------------------
[docs]
class AbstractIssuedCredentialRevocationView(PageContextMixin, DetailView[IssuedCredentialModel]):
"""Revokes a specific issued credential."""
[docs]
http_method_names = ('get', 'post')
[docs]
model = IssuedCredentialModel
[docs]
template_name = 'devices/confirm_revoke.html'
[docs]
context_object_name = 'issued_credential'
[docs]
page_category = DEVICES_PAGE_CATEGORY
[docs]
def get_context_data(self, **kwargs: Any) -> dict[str, Any]:
"""Adds the primary keys to the context.
Args:
kwargs: Keyword arguments are passed to super().get_context_data(**kwargs).
Returns:
The context data.
"""
context = super().get_context_data(**kwargs)
context['revoke_form'] = self.form_class()
context['cert'] = self.object.credential.certificate
context['cred_revoke_url'] = f'{self.page_category}:{self.page_name}_credential_revoke'
return context
[docs]
def post(self, _request: HttpRequest, *_args: Any, **_kwargs: Any) -> HttpResponse:
"""Will try to revoke the requested issued credential.
Args:
request: The Django request object.
*_args: Positional arguments are discarded.
**_kwargs: Keyword arguments are discarded.
Returns:
Redirect to the devices summary.
"""
self.object = self.get_object()
device = self.object.device
if device is None:
raise Http404
reverse_path = reverse(
f'{self.page_category}:{self.page_name}_certificate_lifecycle_management',
kwargs={'pk': device.pk},
)
revoke_form = self.form_class(self.request.POST)
if revoke_form.is_valid():
revocation_reason = revoke_form.cleaned_data['revocation_reason']
cert = self.object.credential.certificate_or_error
status = cert.certificate_status
if status == CertificateModel.CertificateStatus.EXPIRED:
msg = gettext_lazy('Credential is already expired. Cannot revoke expired certificates.')
messages.error(self.request, msg)
return redirect(reverse_path)
if status == CertificateModel.CertificateStatus.REVOKED:
msg = gettext_lazy('Certificate is already revoked. Cannot revoke a revoked certificate again.')
messages.error(self.request, msg)
return redirect(reverse_path)
revoked_successfully, _ = DeviceCredentialRevocation.revoke_certificate(self.object.id, revocation_reason)
if revoked_successfully:
msg = gettext_lazy('Successfully revoked one active credential.')
messages.success(self.request, msg)
else:
messages.error(
self.request, gettext_lazy('Failed to revoke certificate. See logs for more information.')
)
return redirect(reverse_path)
[docs]
class DeviceIssuedCredentialRevocationView(AbstractIssuedCredentialRevocationView):
"""abc."""
[docs]
page_name = DEVICES_PAGE_DEVICES_SUBCATEGORY
[docs]
class OpcUaGdsIssuedCredentialRevocationView(AbstractIssuedCredentialRevocationView):
"""abc."""
[docs]
page_name = DEVICES_PAGE_OPC_UA_SUBCATEGORY
[docs]
class OpcUaGdsPushIssuedCredentialRevocationView(AbstractIssuedCredentialRevocationView):
"""abc."""
[docs]
page_name = DEVICES_PAGE_DEVICES_SUBCATEGORY
[docs]
class AbstractBulkRevokeView(LoggerMixin, PageContextMixin, ListView[DeviceModel]):
"""View to confirm the deletion of multiple Devices."""
[docs]
template_name = 'devices/confirm_bulk_revoke.html'
[docs]
context_object_name = 'devices'
[docs]
queryset: QuerySet[DeviceModel]
[docs]
page_category = DEVICES_PAGE_CATEGORY
[docs]
def get_context_data(self, **kwargs: Any) -> dict[str, Any]:
"""Adds the primary keys to the context.
Args:
kwargs: Keyword arguments are passed to super().get_context_data(**kwargs).
Returns:
The context data.
"""
context = super().get_context_data(**kwargs)
context['pks'] = self.pks
context['revoke_form'] = self.form_class(initial={'pks': self.pks})
context['device_revoke_url'] = f'{self.page_category}:{self.page_name}_device_revoke'
return context
[docs]
def get_queryset(self) -> QuerySet[DeviceModel]:
"""Gets the queryset of DeviceModel objects that are requested to be revoked.
Returns:
The queryset of DeviceModel objects that are requested to be revoked.
"""
if not self.pks:
self.pks = self.kwargs.get('pks', '')
pks_list = get_primary_keys_from_str_as_list_of_ints(pks=self.pks)
qs = DeviceModel.objects.filter(pk__in=pks_list)
found_pks = set(qs.values_list('pk', flat=True))
missing = set(pks_list) - found_pks
self.missing = ', '.join(str(pk) for pk in missing)
return qs
def _set_queryset(self, request: HttpRequest) -> str | None:
try:
self.queryset = self.get_queryset()
except ValueError:
err_msg_template = gettext_lazy('Please select the devices you would like to revoke.')
err_msg = err_msg_template.format(pks=self.pks)
messages.error(request, err_msg)
return 'devices:devices'
except Exception:
err_msg_template = gettext_lazy(
f'Failed to retrieve the queryset for primary keys: {self.pks}.See logs for more details.'
)
err_msg = err_msg_template.format(pks=self.pks)
self.logger.exception(err_msg)
messages.error(request, err_msg)
return 'devices:devices'
if self.missing:
err_msg_template = gettext_lazy(f'Devices for the following primary keys were not found: {self.pks}.')
err_msg = err_msg_template.format(pks=self.missing)
messages.error(request, err_msg)
return 'devices:devices'
return None
[docs]
def get(self, request: HttpRequest, *args: Any, **kwargs: Any) -> HttpResponse:
"""HTTP GET processing.
Args:
request: The Django request object.
*args: Positional arguments passed to super().get().
**kwargs: Keyword arguments passed to super().get().
Returns:
The device deletion view or a redirect to the devices view if one or more pks were not found.
"""
redirect_name = self._set_queryset(request)
if redirect_name:
return redirect(redirect_name)
messages.warning(
request, gettext_lazy('This operation will revoke ALL certificates associated with the selected devices.')
)
return super().get(request, *args, **kwargs)
[docs]
def post(self, request: HttpRequest, *_args: Any, **_kwargs: Any) -> HttpResponse:
"""Will try to revoke all certificate assiciated with the requested DeviceModel records.
Args:
request: The Django request object.
*_args: Positional arguments are discarded.
**_kwargs: Keyword arguments are discarded.
Returns:
Redirect to the devices summary.
"""
revoke_form = self.form_class(self.request.POST)
if revoke_form.is_valid():
self.pks = revoke_form.cleaned_data['pks']
revocation_reason = revoke_form.cleaned_data['revocation_reason']
redirect_name = self._set_queryset(request)
if redirect_name:
return redirect(redirect_name)
now = datetime.datetime.now(datetime.UTC)
issued_credentials_to_revoke_qs = IssuedCredentialModel.objects.filter(
device__in=self.queryset,
credential__certificate__revoked_certificate__isnull=True,
credential__certificate__not_valid_after__gte=now,
)
n_revoked = 0
for credential in issued_credentials_to_revoke_qs:
revoked_successfully, _ = DeviceCredentialRevocation.revoke_certificate(
credential.id, revocation_reason
)
if revoked_successfully:
n_revoked += 1
if n_revoked > 0:
msg = ngettext(
'Successfully revoked one active credential.',
'Successfully revoked %(count)d active credentials.',
n_revoked,
) % {'count': n_revoked}
messages.success(self.request, msg)
else:
messages.error(self.request, gettext_lazy('No credentials were revoked.'))
return redirect('devices:devices')
[docs]
class DeviceBulkRevokeView(AbstractBulkRevokeView):
"""abc."""
[docs]
page_name = DEVICES_PAGE_DEVICES_SUBCATEGORY
[docs]
class OpcUaGdsBulkRevokeView(AbstractBulkRevokeView):
"""abc."""
[docs]
page_name = DEVICES_PAGE_DEVICES_SUBCATEGORY
[docs]
class OpcUaGdsPushBulkRevokeView(AbstractBulkRevokeView):
"""abc."""
[docs]
page_name = DEVICES_PAGE_DEVICES_SUBCATEGORY
[docs]
class AbstractBulkDeleteView(LoggerMixin, PageContextMixin, ListView[DeviceModel]):
"""View to confirm the deletion of multiple Devices."""
[docs]
template_name = 'devices/confirm_delete.html'
[docs]
context_object_name = 'devices'
[docs]
queryset: QuerySet[DeviceModel]
[docs]
page_category = DEVICES_PAGE_CATEGORY
[docs]
def get_context_data(self, **kwargs: Any) -> dict[str, Any]:
"""Adds the primary keys to the context.
Args:
kwargs: Keyword arguments are passed to super().get_context_data(**kwargs).
Returns:
The context data.
"""
context = super().get_context_data(**kwargs)
context['pks'] = self.pks
context['delete_form'] = self.form_class(initial={'pks': self.pks})
context['device_delete_url'] = f'{self.page_category}:{self.page_name}_device_delete'
return context
[docs]
def get_queryset(self) -> QuerySet[DeviceModel]:
"""Gets the queryset of DeviceModel objects that are requested to be deleted.
Returns:
The queryset of DeviceModel objects that are requested to be deleted.
"""
if not self.pks:
self.pks = self.kwargs.get('pks', '')
pks_list = get_primary_keys_from_str_as_list_of_ints(pks=self.pks)
qs = DeviceModel.objects.filter(pk__in=pks_list)
found_pks = set(qs.values_list('pk', flat=True))
missing = set(pks_list) - found_pks
self.missing = ', '.join(str(pk) for pk in missing)
return qs
def _set_queryset(self, request: HttpRequest) -> str | None:
try:
self.queryset = self.get_queryset()
except ValueError:
err_msg_template = gettext_lazy('Please select the devices you would like to delete.')
err_msg = err_msg_template.format(pks=self.pks)
messages.error(request, err_msg)
return 'devices:devices'
except Exception:
err_msg_template = gettext_lazy(
f'Failed to retrieve the queryset for primary keys: {self.pks}.See logs for more details.'
)
err_msg = err_msg_template.format(pks=self.pks)
self.logger.exception(err_msg)
messages.error(request, err_msg)
return 'devices:devices'
if self.missing:
err_msg_template = gettext_lazy(f'Devices for the following primary keys were not found: {self.pks}.')
err_msg = err_msg_template.format(pks=self.missing)
messages.error(request, err_msg)
return 'devices:devices'
return None
[docs]
def get(self, request: HttpRequest, *args: Any, **kwargs: Any) -> HttpResponse:
"""HTTP GET processing.
Args:
request: The Django request object.
*args: Positional arguments passed to super().get().
**kwargs: Keyword arguments passed to super().get().
Returns:
The device deletion view or a redirect to the devices view if one or more pks were not found.
"""
redirect_name = self._set_queryset(request)
if redirect_name:
return redirect(redirect_name)
messages.warning(
request, gettext_lazy('This operation will revoke ALL certificates associated with the selected devices.')
)
return super().get(request, *args, **kwargs)
[docs]
def post(self, request: HttpRequest, *_args: Any, **_kwargs: Any) -> HttpResponse:
"""HTTP POST processing which will try to delete all requested DeviceModel records.
Args:
request: The Django request object.
*_args: Positional arguments are discarded.
**_kwargs: Keyword arguments are discarded.
Returns:
Redirect to the devices summary.
"""
delete_form = self.form_class(self.request.POST)
if delete_form.is_valid():
self.pks = delete_form.cleaned_data['pks']
redirect_name = self._set_queryset(request)
if redirect_name:
return redirect(redirect_name)
try:
count, _ = self.queryset.delete()
success_msg_template = gettext_lazy(
'Successfully deleted {count} devices. All corresponding certificates have been revoked.'
)
success_msg = success_msg_template.format(count=count)
messages.success(request, success_msg)
except Exception:
err_msg = 'Failed to delete DeviceModel records.'
self.logger.exception(err_msg)
messages.error(
request, gettext_lazy('Failed to delete DeviceModel records. See logs for more information.')
)
return redirect('devices:devices')
[docs]
class DeviceBulkDeleteView(AbstractBulkDeleteView):
"""abc."""
[docs]
page_name = DEVICES_PAGE_DEVICES_SUBCATEGORY
[docs]
class OpcUaGdsBulkDeleteView(AbstractBulkDeleteView):
"""abc."""
[docs]
page_name = DEVICES_PAGE_OPC_UA_SUBCATEGORY
[docs]
class OpcUaGdsPushBulkDeleteView(AbstractBulkDeleteView):
"""abc."""
[docs]
page_name = DEVICES_PAGE_DEVICES_SUBCATEGORY
@extend_schema(tags=['Device'])
[docs]
class DeviceViewSet(viewsets.ModelViewSet[DeviceModel]):
"""ViewSet for managing Device instances.
Supports standard CRUD operations such as list, retrieve,
create, update, and delete.
"""
[docs]
queryset = DeviceModel.objects.all()
[docs]
serializer_class = DeviceSerializer
[docs]
action_descriptions: ClassVar[dict[str, str]] = {
'list': 'Retrieve a list of all devices.',
'retrieve': 'Retrieve a single device by id.',
'create': 'Create a new device with name, serial number, and status.',
'update': 'Update an existing device.',
'partial_update': 'Partially update an existing device.',
'destroy': 'Delete a device.',
}
[docs]
def get_view_description(self, html: bool = False) -> str: # noqa: FBT001, FBT002
"""Return a description for the given action."""
if hasattr(self, 'action') and self.action in self.action_descriptions:
return self.action_descriptions[self.action]
return super().get_view_description(html)