Coverage for /home/runner/work/viur-core/viur-core/viur/src/viur/core/bones/uri.py: 76%

1import fnmatch 

2import typing as t 

3from .base import BaseBone, ReadFromClientError, ReadFromClientErrorSeverity 

4from urllib.parse import urlparse, urlunparse 

5from collections.abc import Iterable 

6from collections import namedtuple 

7 

8PORT_MIN: t.Final[int] = 1 

9PORT_MAX: t.Final[int] = 2 ** 16 - 1 

10 

11DEFAULT_PORTS: t.Final[dict[str, int]] = { 

12 "ftp": 21, 

13 "ftps": 990, 

14 "http": 80, 

15 "https": 443, 

16 "sftp": 22, 

17 "ssh": 22, 

18 "ws": 80, 

19 "wss": 443, 

20} 

21"""The port a URL uses when it does not name one, by scheme""" 

22 

23 

24class UriBone(BaseBone): 

25 type = "uri" 

26 

27 def __init__( 

28 self, 

29 *, 

30 accepted_protocols: str | t.Iterable[str] | None = None, 

31 accepted_ports: int | str | t.Iterable[int] | t.Iterable[str] | None = None, 

32 clean_get_params: bool = False, 

33 domain_allowed_list: t.Iterable[str] | None = None, 

34 domain_disallowed_list: t.Iterable[str] | None = None, 

35 local_path_allowed: bool = False, 

36 **kwargs 

37 ): 

38 """ 

39 The UriBone is used for storing URI and URL. 

40 

41 :param accepted_protocols: The accepted protocols can be set to allow only the provide protocols. 

42 :param accepted_ports The accepted ports can be set to allow only the provide ports. 

43 A URL that names no port is checked against the default port of its scheme, 

44 see :data:`DEFAULT_PORTS`. 

45 .. code-block:: python 

46 # Example 

47 UriBone(accepted_ports=1) 

48 UriBone(accepted_ports="2") 

49 UriBone(accepted_ports="1-4") 

50 UriBone(accepted_ports=(1,"2","4-10")) 

51 :param clean_get_params: When set to True, the GET-parameter for the URL will be cleaned. 

52 :param domain_allowed_list: If set, only the URLs that are matched with an entry of this iterable 

53 will be accepted. 

54 :param domain_disallowed_list: If set, only the URLs that are not matched 

55 with an entry of this iterable will be accepted. 

56 :param local_path_allowed: If True, the URLs that are local paths will be prefixed with "/". 

57 """ 

58 super().__init__(**kwargs) 

59 if accepted_ports: 

60 self.accepted_ports = sorted(set(UriBone._build_accepted_ports(accepted_ports)), key=lambda rng: rng.start) 

61 

62 if range(PORT_MIN, PORT_MAX + 1) in self.accepted_ports: 62 ↛ 63line 62 didn't jump to line 63 because the condition on line 62 was never true

63 self.accepted_ports = None # all allowed 

64 else: 

65 self.accepted_ports = None 

66 

67 self.accepted_protocols = accepted_protocols 

68 if self.accepted_protocols: 

69 if isinstance(self.accepted_protocols, str): 

70 # A single protocol, not a set of its characters 

71 self.accepted_protocols = {self.accepted_protocols} 

72 elif isinstance(self.accepted_protocols, Iterable): 

73 self.accepted_protocols = set(self.accepted_protocols) 

74 else: 

75 raise ValueError("accepted_protocols must be a string, an iterable of strings or None") 

76 

77 # Test the wildcard against the normalized set: "*" on its own switches the 

78 # check off, while a pattern such as "http*" is kept and matched by fnmatch. 

79 if "*" in self.accepted_protocols: 

80 self.accepted_protocols = None 

81 

82 if not isinstance(clean_get_params, bool): 82 ↛ 83line 82 didn't jump to line 83 because the condition on line 82 was never true

83 raise ValueError("clean_get_params must be a boolean") 

84 

85 if not isinstance(domain_allowed_list, (list, tuple)) and domain_allowed_list is not None: 85 ↛ 86line 85 didn't jump to line 86 because the condition on line 85 was never true

86 raise ValueError("domain_allowed_list must be a list or a tuple or None") 

87 

88 if not isinstance(domain_disallowed_list, (list, tuple)) and domain_disallowed_list is not None: 88 ↛ 89line 88 didn't jump to line 89 because the condition on line 88 was never true

89 raise ValueError("domain_disallowed_list must be a list or a tuple or None") 

90 

91 if domain_allowed_list is not None: 

92 if any([not isinstance(domain, str) for domain in domain_allowed_list]): 92 ↛ 93line 92 didn't jump to line 93 because the condition on line 92 was never true

93 raise ValueError("domain_allowed_list must only contain strings") 

94 

95 if domain_disallowed_list is not None: 95 ↛ 96line 95 didn't jump to line 96 because the condition on line 95 was never true

96 if any([not isinstance(domain, str) for domain in domain_disallowed_list]): 

97 raise ValueError("domain_disallowed_list must only contain strings") 

98 

99 if domain_allowed_list and domain_disallowed_list: 99 ↛ 100line 99 didn't jump to line 100 because the condition on line 99 was never true

100 raise ValueError("Only one of domain_allowed_list and domain_disallowed_list can be set") 

101 

102 if not isinstance(local_path_allowed, bool): 102 ↛ 103line 102 didn't jump to line 103 because the condition on line 102 was never true

103 raise ValueError("local_path_allowed must be a boolean") 

104 

105 self.clean_get_params = clean_get_params 

106 self.domain_allowed_list = domain_allowed_list 

107 self.domain_disallowed_list = domain_disallowed_list 

108 self.local_path_allowed = local_path_allowed 

109 

110 @classmethod 

111 def _build_accepted_ports(cls, accepted_ports: str | int | t.Iterable[str | int]) -> list[range]: 

112 if isinstance(accepted_ports, str): 

113 if accepted_ports == "*": 113 ↛ 114line 113 didn't jump to line 114 because the condition on line 113 was never true

114 return [range(PORT_MIN, PORT_MAX + 1)] 

115 

116 elif "," in accepted_ports: # list of ranges, values 116 ↛ 117line 116 didn't jump to line 117 because the condition on line 116 was never true

117 return cls._build_accepted_ports([ 

118 value.strip() for value in accepted_ports.split(",") 

119 ]) 

120 

121 elif "-" in accepted_ports: # range of ports 

122 start, end = accepted_ports.split("-", 1) 

123 start = int(start) 

124 end = int(end) 

125 if start > end: 125 ↛ 126line 125 didn't jump to line 126 because the condition on line 125 was never true

126 raise ValueError("Start value must be less than end value") 

127 

128 if start < PORT_MIN: 128 ↛ 129line 128 didn't jump to line 129 because the condition on line 128 was never true

129 raise ValueError("Start value must be greater than zero") 

130 

131 if end > PORT_MAX: 131 ↛ 132line 131 didn't jump to line 132 because the condition on line 131 was never true

132 raise ValueError(f"End value must be less or equal than {PORT_MAX}") 

133 

134 return [range(start, end + 1)] 

135 

136 else: 

137 port = int(accepted_ports) 

138 return [range(port, port + 1)] 

139 

140 elif isinstance(accepted_ports, int): 

141 if accepted_ports < PORT_MIN: 141 ↛ 142line 141 didn't jump to line 142 because the condition on line 141 was never true

142 raise ValueError("Port value must be greater than zero") 

143 

144 if accepted_ports > PORT_MAX: 144 ↛ 145line 144 didn't jump to line 145 because the condition on line 144 was never true

145 raise ValueError(f"Port value must be less or equal than {PORT_MAX}") 

146 

147 return [range(accepted_ports, accepted_ports + 1)] 

148 

149 elif isinstance(accepted_ports, Iterable): 149 ↛ 155line 149 didn't jump to line 155 because the condition on line 149 was always true

150 accepted_ports_value = [] 

151 for accepted_port in accepted_ports: 

152 accepted_ports_value.extend(UriBone._build_accepted_ports(accepted_port)) 

153 return accepted_ports_value 

154 

155 raise ValueError("accepted_ports must be a iterable or an integer or string") 

156 

157 def isInvalid(self, value) -> str | None: 

158 try: 

159 parsed_url = urlparse(value) 

160 except ValueError: 

161 return "Can't parse URL" 

162 

163 if not self.local_path_allowed and parsed_url.scheme == "": 

164 return f"""No protocol specified""" 

165 

166 if self.accepted_ports: 

167 try: 

168 port = parsed_url.port 

169 except ValueError: # the property parses the port, urlparse itself does not 

170 return "Can't read the port from the URL" 

171 

172 if port is None: 

173 # The URL relies on the default of its scheme. An unknown scheme leaves the 

174 # port undecidable, and an undecidable port cannot be an accepted one. 

175 port = DEFAULT_PORTS.get(parsed_url.scheme) 

176 

177 if port is None or not any(port in rng for rng in self.accepted_ports): 

178 return f""""{port}" not in the accepted ports.""" 

179 

180 if self.accepted_protocols: 

181 for protocol in self.accepted_protocols: 

182 if fnmatch.fnmatch(parsed_url.scheme, protocol): 

183 break 

184 else: 

185 return f""""{parsed_url.scheme}" not in the accepted protocols.""" 

186 

187 if self.domain_allowed_list is not None: 

188 if parsed_url.hostname: 188 ↛ 195line 188 didn't jump to line 195 because the condition on line 188 was always true

189 for domain in self.domain_allowed_list: 

190 if fnmatch.fnmatch(parsed_url.hostname, domain) or domain in parsed_url.hostname: 

191 break 

192 else: 

193 return f"""Provided URL is not in the domain allowed list.""" 

194 else: 

195 return f"""Provided URL has no hostname specified.""" 

196 

197 if self.domain_disallowed_list is not None: 197 ↛ 198line 197 didn't jump to line 198 because the condition on line 197 was never true

198 if parsed_url.hostname: 

199 for domain in self.domain_disallowed_list: 

200 if fnmatch.fnmatch(parsed_url.hostname, domain) or domain in parsed_url.hostname: 

201 return f"""Provided URL is in the domain disallowed list.""" 

202 

203 else: 

204 return f"""Provided URL has no hostname specified.""" 

205 

206 def singleValueFromClient(self, value, skel, bone_name, client_data) -> tuple: 

207 if err := self.isInvalid(value): 

208 return value, [ReadFromClientError(ReadFromClientErrorSeverity.Invalid, err)] 

209 

210 parsed_url = urlparse(value) 

211 if self.local_path_allowed and parsed_url.scheme == "": 

212 if value[0] not in "?#/": 

213 value = f"/{value}" 

214 parsed_url = urlparse(value) 

215 

216 if self.clean_get_params: 

217 Components = namedtuple( 

218 typename="Components", 

219 field_names=["scheme", "netloc", "path", "url", "query", "fragment"] 

220 ) 

221 

222 value = urlunparse( 

223 Components( 

224 scheme=parsed_url.scheme, 

225 netloc=parsed_url.netloc, 

226 query=None, # Set the GET-params to None to clear it 

227 path=parsed_url.path, 

228 url=None, 

229 fragment=parsed_url.fragment, 

230 ) 

231 ) 

232 

233 return value, None 

234 

235 def structure(self) -> dict: 

236 return super().structure() | { 

237 "accepted_protocols": list(self.accepted_protocols) if self.accepted_protocols else None, 

238 "accepted_ports": [(rng.start, rng.stop) for rng in self.accepted_ports] if self.accepted_ports else None, 

239 "clean_get_params": self.clean_get_params, 

240 "domain_allowed_list": self.domain_allowed_list, 

241 "domain_disallowed_list": self.domain_disallowed_list, 

242 "local_path_allowed": self.local_path_allowed, 

243 }