| 845 | |
| 846 | |
| 847 | class IPField(Field[Union[str, Net], bytes]): |
| 848 | def __init__(self, name, default): |
| 849 | # type: (str, Optional[str]) -> None |
| 850 | Field.__init__(self, name, default, "4s") |
| 851 | |
| 852 | def h2i(self, pkt, x): |
| 853 | # type: (Optional[Packet], Union[AnyStr, List[AnyStr]]) -> Any |
| 854 | if isinstance(x, bytes): |
| 855 | x = plain_str(x) # type: ignore |
| 856 | if isinstance(x, _ScopedIP): |
| 857 | return x |
| 858 | elif isinstance(x, str): |
| 859 | x = ScopedIP(x) |
| 860 | try: |
| 861 | inet_aton(x) |
| 862 | except socket.error: |
| 863 | return Net(x) |
| 864 | elif isinstance(x, tuple): |
| 865 | if len(x) != 2: |
| 866 | raise ValueError("Invalid IP format") |
| 867 | return Net(*x) |
| 868 | elif isinstance(x, list): |
| 869 | return [self.h2i(pkt, n) for n in x] |
| 870 | return x |
| 871 | |
| 872 | def i2h(self, pkt, x): |
| 873 | # type: (Optional[Packet], Optional[Union[str, Net]]) -> str |
| 874 | return cast(str, x) |
| 875 | |
| 876 | def resolve(self, x): |
| 877 | # type: (str) -> str |
| 878 | if self in conf.resolve: |
| 879 | try: |
| 880 | ret = socket.gethostbyaddr(x)[0] |
| 881 | except Exception: |
| 882 | pass |
| 883 | else: |
| 884 | if ret: |
| 885 | return ret |
| 886 | return x |
| 887 | |
| 888 | def i2m(self, pkt, x): |
| 889 | # type: (Optional[Packet], Optional[Union[str, Net]]) -> bytes |
| 890 | if x is None: |
| 891 | return b'\x00\x00\x00\x00' |
| 892 | return inet_aton(plain_str(x)) |
| 893 | |
| 894 | def m2i(self, pkt, x): |
| 895 | # type: (Optional[Packet], bytes) -> str |
| 896 | return inet_ntoa(x) |
| 897 | |
| 898 | def any2i(self, pkt, x): |
| 899 | # type: (Optional[Packet], Any) -> Any |
| 900 | return self.h2i(pkt, x) |
| 901 | |
| 902 | def i2repr(self, pkt, x): |
| 903 | # type: (Optional[Packet], Union[str, Net]) -> str |
| 904 | if isinstance(x, _ScopedIP) and x.scope: |
no outgoing calls
no test coverage detected