Coverage for src/lib/contact.py: 97%

1 

2from socket import getaddrinfo, gaierror 

3from ipaddress import ip_address 

4 

5 

6class Contact: 

7 addr: str = None 

8 port: int = None 

9 is_valid: bool = False 

10 is_ipv6: bool = False 

11 

12 def __init__(self) -> None: 

13 pass 

14 

15 def __str__(self) -> str: 

16 return f'{self.addr}:{self.port}' 

17 

18 @staticmethod 

19 def parse(raw: str) -> 'Contact': 

20 contact = Contact() 

21 

22 if '[' in raw and ']' in raw: 

23 # IPv6 

24 items = raw.split(']') 

25 items = [ 

26 items[0][1:], 

27 int(items[1][1:]), 

28 ] 

29 else: 

30 items = raw.split(':') 

31 

32 items_len = len(items) 

33 

34 if items_len == 1: 

35 contact.addr = items[0] 

36 contact.port = None 

37 elif items_len == 2: 

38 contact.addr = items[0] 

39 if items[1] == '': 

40 contact.port = None 

41 else: 

42 contact.port = int(items[1]) 

43 elif items_len > 2: 

44 # IPv6 

45 contact.addr = ':'.join(items[0:-1]) 

46 contact.port = int(items[-1]) 

47 contact.is_ipv6 = True 

48 

49 if contact.addr == '': 

50 contact.addr = 'private' 

51 

52 return contact 

53 

54 @staticmethod 

55 def resolve(raw: str, raddr: str = None) -> 'Contact': 

56 contact = Contact.parse(raw) 

57 

58 if contact.addr == 'public': 

59 contact.addr = raddr 

60 elif contact.addr == 'private': 

61 contact.addr = None 

62 contact.port = None 

63 else: 

64 try: 

65 ip_add = str(ip_address(contact.addr)) 

66 if ip_add[0:4] == '127.' or ip_add[0:4] == '0.0.' or ip_add == '::1': 

67 # Localhost is invalid. 

68 contact.addr = None 

69 except ValueError: 

70 # Contact is hostname 

71 try: 

72 results = getaddrinfo(contact.addr, None) 

73 for result in results: 

74 ip_add = result[4][0] 

75 if ip_add[0:4] == '127.' or ip_add[0:4] == '0.0.' or ip_add == '::1': 

76 # Localhost is invalid. 

77 contact.addr = None 

78 break 

79 except gaierror: 

80 contact.addr = None 

81 

82 contact.is_valid = contact.addr is not None and contact.port is not None 

83 return contact