Coverage for ipam/models/services.py: 60%
80 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 18:35 +0000
1from django.contrib.contenttypes.fields import GenericForeignKey
2from django.contrib.postgres.fields import ArrayField
3from django.contrib.postgres.indexes import GinIndex
4from django.core.exceptions import ValidationError
5from django.db import models
6from django.utils.translation import gettext_lazy as _
8from ipam.choices import *
9from ipam.constants import *
10from ipam.utils import group_port_mappings, legacy_protocol_and_ports, split_port_mapping
11from ipam.validators import validate_port_mappings
12from netbox.models import PrimaryModel
13from netbox.models.features import ContactsMixin
14from utilities.data import array_to_ranges
16__all__ = (
17 'Service',
18 'ServiceTemplate',
19)
21# Fixed protocol value -> label map, built once (the choice set is static per process) rather than
22# rebuilt on every port_mappings_list render.
23SERVICE_PROTOCOL_LABELS = dict(ServiceProtocolChoices)
26class ServiceBase(models.Model):
27 """
28 Shared behavior for Service and ServiceTemplate. Protocol/port data is stored as a single array of
29 ``protocol/port`` strings (e.g. ``['tcp/80', 'tcp/443', 'udp/53']``), allowing a service to expose
30 the same port on multiple protocols.
31 """
32 port_mappings = ArrayField(
33 base_field=models.CharField(max_length=63),
34 verbose_name=_('port mappings'),
35 help_text=_("Protocol/port pairs, e.g. tcp/80"),
36 blank=True,
37 default=list,
38 )
40 class Meta:
41 abstract = True
43 def __str__(self):
44 return self.name
46 def clean(self):
47 super().clean()
48 # validate_port_mappings returns the canonical form (integer ports), so storing its result
49 # normalizes any entry that bypassed the form field (e.g. a raw REST payload of 'tcp/080'). Key
50 # its errors to the field — it raises unkeyed, which full_clean() would otherwise file as a
51 # non-field (__all__) error rather than against port_mappings.
52 try:
53 self.port_mappings = validate_port_mappings(self.port_mappings)
54 except ValidationError as e:
55 raise ValidationError({'port_mappings': e.messages})
56 if not self.port_mappings:
57 raise ValidationError({'port_mappings': _("At least one port mapping is required.")})
59 @staticmethod
60 def _normalize_mapping(mapping):
61 # Normalize a stored/incoming mapping's port to an integer so a non-canonical value (e.g. a
62 # raw-DB 'tcp/080') compares equal to its canonical form ('tcp/80').
63 protocol, port = split_port_mapping(mapping)
64 return f'{protocol}/{int(port)}' if port.isdigit() else mapping
66 def _add_port_mappings(self, mappings):
67 """
68 Add the given canonical ``protocol/port`` strings to ``port_mappings``, skipping any already
69 present (matched by normalized form). The merged list is left for ``clean()`` to validate.
71 Internal helper called from the Service/ServiceTemplate bulk-edit view's pre_save_operations()
72 hook, so the merge is part of the single bulk-edit save (one change-log entry) and the model
73 stays unaware of the bulk-edit form. Underscore-prefixed to keep it out of the way of the
74 identically-named ``add_port_mappings`` bulk-edit form field (which the generic bulk-edit view
75 assigns onto the object).
76 """
77 existing = {self._normalize_mapping(mapping) for mapping in self.port_mappings}
78 self.port_mappings = list(self.port_mappings) + [
79 mapping for mapping in mappings if self._normalize_mapping(mapping) not in existing
80 ]
82 def _remove_port_mappings(self, mappings):
83 """
84 Remove the given canonical ``protocol/port`` strings from ``port_mappings`` (matched by
85 normalized form). The result is left for ``clean()`` to validate (range, duplicates, and the
86 at-least-one rule). Internal helper called from the bulk-edit view's pre_save_operations() hook
87 (see ``_add_port_mappings``).
88 """
89 remove = {self._normalize_mapping(mapping) for mapping in mappings}
90 self.port_mappings = [
91 mapping for mapping in self.port_mappings if self._normalize_mapping(mapping) not in remove
92 ]
94 @property
95 def port_mappings_list(self):
96 """
97 Return a user-friendly list of port mappings, collapsing consecutive ports within a protocol into
98 a range, e.g. "TCP/80, TCP/443, UDP/53" or "TCP/8000-8100".
99 """
100 parts = []
101 for protocol, ports in group_port_mappings(self.port_mappings).items():
102 label = SERVICE_PROTOCOL_LABELS.get(protocol, protocol)
103 int_ports = [int(port) for port in ports if port.isdigit()]
104 for port_range in array_to_ranges(int_ports):
105 if len(port_range) == 1:
106 parts.append(f'{label}/{port_range[0]}')
107 else:
108 parts.append(f'{label}/{port_range[0]}-{port_range[1]}')
109 # A port that isn't a plain integer is only reachable via a write that bypassed validation;
110 # render it verbatim rather than raising, matching sorted_int_ports and normalize_port_mapping.
111 parts.extend(f'{label}/{port}' for port in ports if not port.isdigit())
112 return ', '.join(parts)
114 # Read-only legacy accessors mirroring the deprecated REST/GraphQL protocol/ports fields, retained
115 # for backward compatibility with code that read the old single-protocol fields. A multi-protocol
116 # service has no single-protocol form, so both return None (ports=[] when there are no mappings).
117 # TODO: Remove these in v5.0 once backward compatibility is dropped.
118 @property
119 def _legacy_protocol_ports(self):
120 # Recomputed on access (grouping a handful of strings is cheap) rather than cached, so a mutation
121 # of port_mappings — e.g. via _add_port_mappings()/_remove_port_mappings() — is always reflected
122 # by the protocol/ports accessors, with no cache to invalidate.
123 return legacy_protocol_and_ports(self.port_mappings)
125 # Return types are annotated so drf-spectacular can resolve these properties when it builds the
126 # write-side serializer schema (without them it warns and falls back to string).
127 @property
128 def protocol(self) -> str | None:
129 return self._legacy_protocol_ports[0]
131 @property
132 def ports(self) -> list[int] | None:
133 return self._legacy_protocol_ports[1]
136class ServiceTemplate(ServiceBase, PrimaryModel):
137 """
138 A template for a Service to be applied to a device or virtual machine.
139 """
140 name = models.CharField(
141 verbose_name=_('name'),
142 max_length=100,
143 unique=True
144 )
146 clone_fields = ('port_mappings', 'description')
148 class Meta:
149 indexes = (
150 GinIndex(fields=('port_mappings',)),
151 )
152 ordering = ('name',)
153 verbose_name = _('application service template')
154 verbose_name_plural = _('application service templates')
157class Service(ContactsMixin, ServiceBase, PrimaryModel):
158 """
159 A Service represents a layer-four service (e.g. HTTP or SSH) running on a Device or VirtualMachine. A Service may
160 optionally be tied to one or more specific IPAddresses belonging to its parent.
161 """
162 parent_object_type = models.ForeignKey(
163 to='contenttypes.ContentType',
164 on_delete=models.PROTECT,
165 related_name='+',
166 )
167 parent_object_id = models.PositiveBigIntegerField()
168 parent = GenericForeignKey(
169 ct_field='parent_object_type',
170 fk_field='parent_object_id'
171 )
172 name = models.CharField(
173 max_length=100,
174 verbose_name=_('name')
175 )
176 ipaddresses = models.ManyToManyField(
177 to='ipam.IPAddress',
178 related_name='services',
179 blank=True,
180 verbose_name=_('IP addresses'),
181 help_text=_("The specific IP addresses (if any) to which this application service is bound")
182 )
184 clone_fields = (
185 'port_mappings', 'description', 'parent', 'ipaddresses',
186 )
188 class Meta:
189 indexes = (
190 models.Index(fields=('name', 'id')), # Default ordering
191 models.Index(fields=('parent_object_type', 'parent_object_id')),
192 GinIndex(fields=('port_mappings',)),
193 )
194 ordering = ('name', 'id')
195 verbose_name = _('application service')
196 verbose_name_plural = _('application services')