Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 10 additions & 8 deletions pcapkit/corekit/fields/ipaddress.py
Original file line number Diff line number Diff line change
Expand Up @@ -242,9 +242,10 @@ def pre_process(self, value: 'IPv4Interface | bytes | int | str', packet: 'dict[
val = value
else:
with _reraise_as_field_value_error('invalid IP interface'):
val = ipaddress.ip_interface(value) # type: ignore[assignment]
if val.version != self.version:
raise FieldValueError(f'IP version mismatch: {val.version} != {self.version}')
parsed = ipaddress.ip_interface(value)
if not isinstance(parsed, ipaddress.IPv4Interface):
raise FieldValueError(f'IP version mismatch: {parsed.version} != {self.version}')
val = parsed

ip = val.ip
mask = val.netmask
Expand Down Expand Up @@ -281,7 +282,7 @@ def post_process(self, value: 'bytes', packet: 'dict[str, Any]') -> 'IPv4Interfa

with _reraise_as_field_value_error('invalid IPv4 interface'):
val = ipaddress.ip_interface(f'{ip}/{mask}')
if val.version != self.version:
if not isinstance(val, ipaddress.IPv4Interface):
raise FieldValueError(f'IP version mismatch: {val.version} != {self.version}')
return val

Expand Down Expand Up @@ -326,9 +327,10 @@ def pre_process(self, value: 'IPv6Interface | bytes | int | str', packet: 'dict[
val = value
else:
with _reraise_as_field_value_error('invalid IP interface'):
val = ipaddress.ip_interface(value) # type: ignore[assignment]
if val.version != self.version:
raise FieldValueError(f'IP version mismatch: {val.version} != {self.version}')
parsed = ipaddress.ip_interface(value)
if not isinstance(parsed, ipaddress.IPv6Interface):
raise FieldValueError(f'IP version mismatch: {parsed.version} != {self.version}')
val = parsed

ip = val.ip
prefixlen = cast('int', val._prefixlen) # type: ignore[attr-defined] # pylint: disable=protected-access
Expand Down Expand Up @@ -372,6 +374,6 @@ def post_process(self, value: 'bytes', packet: 'dict[str, Any]') -> 'IPv6Interfa

with _reraise_as_field_value_error('invalid IPv6 interface'):
val = ipaddress.ip_interface(f'{ip}/{prefixlen}')
if val.version != self.version:
if not isinstance(val, ipaddress.IPv6Interface):
raise FieldValueError(f'IP version mismatch: {val.version} != {self.version}')
return val
Loading