Coverage for gws-app/gws/lib/xmlx/validator.py: 77%

1"""Schema validator.""" 

2 

3import re 

4import os 

5import lxml.etree 

6import requests 

7 

8import gws 

9 

10from . import util 

11 

12 

13class Error(gws.Error): 

14 def __init__(self, *args, **kwargs): 

15 super().__init__(*args, **kwargs) 

16 self.message = args[0] 

17 self.lineno = args[1] 

18 

19 

20def validate(xml: str | bytes): 

21 try: 

22 parser = lxml.etree.XMLParser(resolve_entities=False, no_network=True) 

23 parser.resolvers.add(_CachingResolver()) 

24 

25 schema_locations = _extract_schema_locations(xml) 

26 xsd = _create_combined_xsd(schema_locations) 

27 

28 xml_tree = _etree(xml, parser) 

29 schema_tree = _etree(xsd, parser) 

30 schema = lxml.etree.XMLSchema(schema_tree) 

31 except lxml.etree.Error as exc: 

32 raise _error(exc) from exc 

33 

34 try: 

35 schema.assertValid(xml_tree) 

36 return True 

37 except Exception as exc: 

38 raise _error(exc) from exc 

39 

40 

41def _extract_schema_locations(xml: str | bytes) -> dict: 

42 tree = _etree(xml, None) 

43 root = tree.getroot() 

44 

45 xsi_ns = '{http://www.w3.org/2001/XMLSchema-instance}' 

46 attr = root.get(f'{xsi_ns}schemaLocation') 

47 if not attr: 

48 attr = root.get('schemaLocation') 

49 if not attr: 

50 return {} 

51 

52 d = {} 

53 

54 parts = attr.strip().split() 

55 while parts: 

56 namespace = parts.pop(0) 

57 location = parts.pop(0) 

58 d[namespace] = location 

59 

60 return d 

61 

62 

63def _create_combined_xsd(schema_locations: dict) -> str: 

64 xml = [] 

65 xml.append('<?xml version="1.0" encoding="UTF-8"?>') 

66 xml.append('<xs:schema xmlns:xs="http://www.w3.org/2001/XMLSchema">') 

67 

68 for ns, loc in schema_locations.items(): 

69 xml.append(f'<xs:import namespace="{util.escape_attribute(ns)}" schemaLocation="{util.escape_attribute(loc)}"/>') 

70 

71 xml.append('</xs:schema>\n') 

72 

73 return '\n'.join(xml) 

74 

75 

76def _etree(xml: str | bytes, parser: lxml.etree.XMLParser | None) -> lxml.etree.ElementTree: 

77 if isinstance(xml, str): 

78 xml = xml.encode('utf-8') 

79 return lxml.etree.ElementTree(lxml.etree.fromstring(xml, parser)) 

80 

81 

82def _error(exc): 

83 # exc is either {'message': ..., 'lineno': ...} 

84 # or {'error_log': '<string>:17:0:ERROR:...} 

85 

86 cls = exc.__class__.__name__ 

87 

88 s = getattr(exc, 'error_log', None) 

89 if s: 

90 try: 

91 lineno = int(s.split(':')[1]) 

92 except Exception: 

93 lineno = 0 

94 return Error(f'{cls}: {s}', lineno) 

95 

96 lineno = getattr(exc, 'lineno', 0) 

97 return Error(f'{cls}: {exc}', lineno) 

98 

99 

100class _CachingResolver(lxml.etree.Resolver): 

101 def resolve(self, url, id, context): 

102 if url.startswith(('http://', 'https://')): 

103 if '.loc' in url or 'local' in url: 

104 buf = _download_url(url, with_cache=False) 

105 else: 

106 buf = _download_url(url, with_cache=True) 

107 return self.resolve_string(buf, context, base_url=url) 

108 

109 return super().resolve(url, id, context) 

110 

111 

112def _download_url(url: str, with_cache: bool) -> bytes: 

113 if not with_cache: 

114 return _raw_download_url(url) 

115 

116 cache_dir = gws.u.ensure_dir(gws.c.CACHE_DIR + '/xmlx') 

117 cache_path = _cache_path(cache_dir, url) 

118 

119 if os.path.exists(cache_path): 

120 return gws.u.read_file_b(cache_path) 

121 

122 content = _raw_download_url(url) 

123 gws.u.write_file_b(cache_path, content) 

124 return content 

125 

126 

127def _raw_download_url(url: str) -> bytes: 

128 gws.log.debug(f'xmlx.validator: downloading {url!r}') 

129 response = requests.get(url, timeout=10) 

130 if response.status_code != 200: 

131 raise ValueError(f'Failed to download {url!r}: {response.status_code}') 

132 return response.content 

133 

134 

135def _cache_path(cache_dir: str, url: str) -> str: 

136 u = url.strip().split('//')[-1] 

137 if '?' in u: 

138 u = u.split('?', 1)[0] 

139 fname = 'index.xml' 

140 parts = u.split('/') 

141 

142 if u.endswith('/'): 

143 parts.pop() 

144 else: 

145 m = re.search(r'[^/]+\.[a-z]+$', parts[-1]) 

146 if m: 

147 fname = m.group(0) 

148 parts.pop() 

149 

150 d = '/'.join(_to_dirname(p) for p in parts) 

151 if not d: 

152 return cache_dir + '/' + fname 

153 d = gws.u.ensure_dir(cache_dir + '/' + d) 

154 return d + '/' + fname 

155 

156 

157def _to_dirname(s: str) -> str: 

158 s = s.lower().strip().lstrip('.') 

159 s = re.sub(r'[^a-zA-Z0-9.]+', '_', s).strip('_') 

160 return s