Coverage for pass_import/audit.py: 100%

1# -*- encoding: utf-8 -*- 

2# pass import - Passwords importer swiss army knife 

3# Copyright (C) 2017-2024 Alexandre PUJOL <alexandre@pujol.io>. 

4# 

5 

6import hashlib 

7 

8import requests 

9from zxcvbn import zxcvbn 

10from typing import List, Tuple 

11 

12import pass_import 

13 

14 

15class PwnedAPI(): 

16 """Simple wrapper for https://haveibeenpwned.com API.""" 

17 

18 def __init__(self): 

19 self.headers = { 

20 'user-agent': f"{pass_import.__title__}/{pass_import.__version__}"} 

21 

22 def password_range(self, prefix: str) -> Tuple[List[str], List[int]]: 

23 """Query the haveibeenpwned api to retrieve the bucket ``prefix``.""" 

24 url = f"https://api.pwnedpasswords.com/range/{prefix}" 

25 res = requests.get(url, headers=self.headers, verify=True, timeout=5) 

26 res.raise_for_status() 

27 

28 hashes = [] 

29 counts = [] 

30 for item in res.text.split('\r\n'): 

31 partialhash, count = item.split(':') 

32 hashes.append(prefix + partialhash) 

33 counts.append(int(count)) 

34 return (hashes, counts) 

35 

36 

37class Audit(): 

38 """Audit passwords for vulnerabilities. 

39 

40 Based on the PassAudit class from pass-audit. 

41 See https://github.com/roddhjav/pass-audit for more information. 

42 

43 :param list[dict] data: The list of password entries to audit 

44 Each password entry is a dictionary. 

45 

46 """ 

47 

48 def __init__(self, data): 

49 self.data = data 

50 self.breached = [] 

51 self.weak = [] 

52 self.duplicated = [] 

53 

54 @property 

55 def report(self): 

56 """Get audit result.""" 

57 return { 

58 'breached': self.breached, 

59 'weak': self.weak, 

60 'duplicated': self.duplicated 

61 } 

62 

63 def password(self): 

64 """K-anonimity password breach detection on haveibeenpwned.com.""" 

65 # Generate the list of hashes and prefixes to query. 

66 data = [] 

67 api = PwnedAPI() 

68 buckets = {} 

69 for entry in self.data: 

70 if entry.get('password', '') == '': 

71 continue 

72 password = entry['password'].encode("utf8") 

73 phash = hashlib.sha1(password).hexdigest().upper() # nosec 

74 prefix = phash[0:5] 

75 data.append((entry, phash, prefix)) 

76 if prefix not in buckets: 

77 buckets[prefix] = api.password_range(prefix) 

78 

79 # Compare the data and return the breached passwords. 

80 for entry, phash, prefix in data: 

81 if phash in buckets[prefix][0]: 

82 index = buckets[prefix][0].index(phash) 

83 count = buckets[prefix][1][index] 

84 self.breached.append((entry.get('password', ''), count)) 

85 

86 def zxcvbn(self): 

87 """Password strength estimaton usuing Dropbox' zxcvbn.""" 

88 for entry in self.data: 

89 if entry.get('password', '') == '': 

90 continue 

91 password = entry['password'] 

92 user_input = list(entry.values()) 

93 if password in user_input: 

94 user_input.remove(password) 

95 results = zxcvbn(password, user_inputs=user_input) 

96 if results['score'] <= 2: 

97 self.weak.append((password, results)) 

98 

99 def duplicates(self): 

100 """Check for duplicated passwords.""" 

101 seen = {} 

102 for entry in self.data: 

103 if entry.get('password', '') == '': 

104 continue 

105 password = entry['password'] 

106 if password in seen: 

107 seen[password].append(entry) 

108 else: 

109 seen[password] = [entry] 

110 

111 for entries in seen.values(): 

112 if len(entries) > 1: 

113 self.duplicated.append(entries)