Coverage for ipam/validators.py: 56%

42 statements  

« prev     ^ index     » next       coverage.py v7.15.2, created at 2026-10-10 18:35 +0000

1from django.core.exceptions import ValidationError 

2from django.core.validators import BaseValidator, RegexValidator 

3from django.utils.translation import gettext_lazy as _ 

4 

5 

6def validate_port_mappings(mappings): 

7 """ 

8 Validate a list of service port mappings, i.e. ``protocol/port`` strings such as ``'tcp/80'``. 

9 Ensures each entry is well-formed, uses a known protocol, falls within the permitted port range, 

10 and is not duplicated. Raises a ``ValidationError`` describing the first problem found. 

11 

12 Returns the list in a canonical, normalized form (integer ports, so ``'tcp/080'`` becomes 

13 ``'tcp/80'``, and the protocol lowercased, so ``'TCP/80'`` becomes ``'tcp/80'``); callers should 

14 persist the returned value so every entry path stores identical strings and remains matchable by the 

15 port filters. Protocol matching is case-insensitive, so all paths (REST, CSV import, model form) 

16 accept any case without each having to fold it first. Shared by the model (``ServiceBase.clean()``), 

17 the model form field (``PortMappingField``), the CSV import form, and the REST API serializers so all 

18 paths enforce identical rules. 

19 """ 

20 # Imported lazily to avoid a circular import during settings load (this module is imported by 

21 # ipam.models, and ipam.constants pulls in ipam.choices, which reads settings.FIELD_CHOICES). 

22 from ipam.choices import ServiceProtocolChoices 

23 from ipam.constants import SERVICE_PORT_MAX, SERVICE_PORT_MIN 

24 from ipam.utils import split_port_mapping 

25 

26 # A set, since this is consulted once per mapping and a service may define thousands 

27 valid_protocols = set(ServiceProtocolChoices.values()) 

28 seen = set() 

29 normalized_mappings = [] 

30 for mapping in mappings: 

31 protocol, port = split_port_mapping(mapping) 

32 if not port: 32 ↛ 39line 32 didn't jump to line 39 because the condition on line 32 was always true

33 raise ValidationError( 

34 _("Invalid port mapping '{mapping}'. Expected format protocol/port (e.g. tcp/80).").format( 

35 mapping=mapping 

36 ) 

37 ) 

38 # The error reports the protocol as supplied rather than the folded form. 

39 if (canonical_protocol := protocol.lower()) not in valid_protocols: 

40 raise ValidationError(_("Invalid protocol: {protocol}").format(protocol=protocol)) 

41 try: 

42 port_number = int(port) 

43 except ValueError: 

44 raise ValidationError(_("Invalid port number: {port}").format(port=port)) 

45 if not SERVICE_PORT_MIN <= port_number <= SERVICE_PORT_MAX: 

46 raise ValidationError( 

47 _("Port {port} is not within the permitted range ({min}-{max}).").format( 

48 port=port_number, min=SERVICE_PORT_MIN, max=SERVICE_PORT_MAX 

49 ) 

50 ) 

51 # Normalize the port to an integer so e.g. tcp/80 and tcp/080 count as duplicates and are 

52 # stored identically (leaving the raw string would make tcp/080 invisible to the port filter). 

53 normalized = f'{canonical_protocol}/{port_number}' 

54 if normalized in seen: 

55 raise ValidationError(_("Duplicate port mapping: {mapping}").format(mapping=mapping)) 

56 seen.add(normalized) 

57 normalized_mappings.append(normalized) 

58 

59 return normalized_mappings 

60 

61 

62def prefix_validator(prefix): 

63 if prefix.ip != prefix.cidr.ip: 63 ↛ 64line 63 didn't jump to line 64 because the condition on line 63 was never true

64 raise ValidationError( 

65 _("{prefix} is not a valid prefix. Did you mean {suggested}?").format( 

66 prefix=prefix, suggested=prefix.cidr 

67 ) 

68 ) 

69 

70 

71class MaxPrefixLengthValidator(BaseValidator): 

72 message = _('The prefix length must be less than or equal to %(limit_value)s.') 

73 code = 'max_prefix_length' 

74 

75 def compare(self, a, b): 

76 return a.prefixlen > b 

77 

78 

79class MinPrefixLengthValidator(BaseValidator): 

80 message = _('The prefix length must be greater than or equal to %(limit_value)s.') 

81 code = 'min_prefix_length' 

82 

83 def compare(self, a, b): 

84 return a.prefixlen < b 

85 

86 

87DNSValidator = RegexValidator( 

88 regex=r'^([0-9A-Za-z_-]+|\*)(\.[0-9A-Za-z_-]+)*\.?$', 

89 message=_('Only alphanumeric characters, asterisks, hyphens, periods, and underscores are allowed in DNS names'), 

90 code='invalid' 

91)