Coverage for product_risk_suite/scraper/models.py: 91%

1from cryptography.fernet import Fernet, InvalidToken 

2from django.conf import settings 

3from django.core.exceptions import ValidationError 

4from django.db import models 

5 

6 

7def _fernet(): 

8 return Fernet(settings.EVIDENCE_SCRAPER_CREDENTIAL_KEY.encode()) 

9 

10 

11class RemoteCredential(models.Model): 

12 BASIC = "basic" 

13 TOKEN = "token" 

14 AUTH_TYPE_CHOICES = { 

15 BASIC: "Username / password", 

16 TOKEN: "API token", 

17 } 

18 

19 name = models.CharField(max_length=100, unique=True, help_text="Human identifier, referenced by Scrapers.") 

20 url = models.URLField(help_text="Base URL of the remote system, e.g. https://openproject.example.com") 

21 auth_type = models.CharField(max_length=10, choices=AUTH_TYPE_CHOICES, default=BASIC) 

22 username = models.CharField(max_length=150, blank=True) 

23 

24 _password_enc = models.CharField(max_length=500, blank=True, db_column="password_enc") 

25 _api_token_enc = models.CharField(max_length=500, blank=True, db_column="api_token_enc") 

26 

27 def __str__(self): 

28 return self.name 

29 

30 def clean(self): 

31 if self.auth_type == self.BASIC and not (self.username and self.password): 

32 raise ValidationError("Username and password are required for basic auth credentials.") 

33 if self.auth_type == self.TOKEN and not self.api_token: 

34 raise ValidationError("An API token is required for token auth credentials.") 

35 

36 @property 

37 def password(self): 

38 return self._decrypt(self._password_enc) 

39 

40 @password.setter 

41 def password(self, raw_value): 

42 self._password_enc = self._encrypt(raw_value) 

43 

44 @property 

45 def api_token(self): 

46 return self._decrypt(self._api_token_enc) 

47 

48 @api_token.setter 

49 def api_token(self, raw_value): 

50 self._api_token_enc = self._encrypt(raw_value) 

51 

52 @staticmethod 

53 def _encrypt(raw_value): 

54 if not raw_value: 

55 return "" 

56 return _fernet().encrypt(raw_value.encode()).decode() 

57 

58 @staticmethod 

59 def _decrypt(enc_value): 

60 if not enc_value: 

61 return "" 

62 try: 

63 return _fernet().decrypt(enc_value.encode()).decode() 

64 except InvalidToken: 

65 return "" 

66 

67 

68class Scraper(models.Model): 

69 JSON = "json" 

70 HTML = "html" 

71 RESPONSE_FORMAT_CHOICES = { 

72 JSON: "JSON", 

73 HTML: "HTML", 

74 } 

75 

76 name = models.CharField(max_length=100, unique=True) 

77 credential = models.ForeignKey( 

78 RemoteCredential, null=True, blank=True, on_delete=models.PROTECT, related_name="scrapers", 

79 help_text="Leave blank for publicly accessible endpoints that need no authentication.", 

80 ) 

81 response_format = models.CharField(max_length=10, choices=RESPONSE_FORMAT_CHOICES, default=JSON) 

82 

83 assignee_path = models.CharField( 

84 max_length=255, 

85 help_text="JSON: dotted/indexed path, e.g. 'assignee.name' or 'fields[0].assignee'. " 

86 "HTML: a CSS selector, optionally suffixed '::attr(name)' to read an attribute instead of text.", 

87 ) 

88 status_path = models.CharField( 

89 max_length=255, 

90 help_text="Same syntax as assignee path. The extracted raw value is mapped to a local " 

91 "Status via the mappings below.", 

92 ) 

93 extra_headers = models.JSONField(default=dict, blank=True, help_text="Optional extra HTTP headers.") 

94 

95 def __str__(self): 

96 return self.name 

97 

98 

99class ScraperStatusMapping(models.Model): 

100 scraper = models.ForeignKey(Scraper, on_delete=models.CASCADE, related_name="status_mappings") 

101 remote_value = models.CharField(max_length=255, help_text="Exact status value as returned by the remote system.") 

102 status = models.ForeignKey("risk_assessment.Status", on_delete=models.PROTECT) 

103 

104 class Meta: 

105 unique_together = ("scraper", "remote_value") 

106 

107 def __str__(self): 

108 return f"{self.remote_value} -> {self.status}" 

109 

110 

111class AssigneeMapping(models.Model): 

112 base_url = models.URLField( 

113 help_text="Base URL of the remote system, e.g. https://codeberg.org. Matched as a prefix " 

114 "against the scraped URL.", 

115 ) 

116 remote_username = models.CharField( 

117 max_length=150, help_text="Username as returned by the remote system, e.g. CMon.", 

118 ) 

119 user = models.ForeignKey( 

120 settings.AUTH_USER_MODEL, on_delete=models.CASCADE, related_name="scraper_assignee_mappings", 

121 ) 

122 

123 class Meta: 

124 unique_together = ("base_url", "remote_username") 

125 

126 def __str__(self): 

127 return f"{self.remote_username} @ {self.base_url} -> {self.user}"