Coverage for creepo/pipproxy.py: 72%

1"""The pip proxy""" 

2from urllib.parse import urlparse 

3import lxml.etree as ET 

4 

5import cherrypy 

6 

7from httpproxy import HttpProxy 

8 

9 

10class PipProxy: 

11 """The pip proxy""" 

12 

13 def __init__(self, config): 

14 self.logger = config['logger'] 

15 self.config = config 

16 self.key = 'pip' 

17 if self.key not in self.config: 

18 self.config[self.key] = { 

19 'registry': 'https://pypi.org/simple', 'self': 'https://localhost:4443/pip'} 

20 

21 self._proxy = HttpProxy(self.config, self.key) 

22 self.logger.debug('PipProxy instantiated with %s', 

23 self.config[self.key]) 

24 

25 def callback(self, _input_bytes, request): 

26 """write the file to disk""" 

27 parser = ET.XMLParser(recover=True) 

28 doc = _input_bytes 

29 tree = ET.fromstring(doc, parser=parser) 

30 

31 if tree is not None: 

32 for node in tree.findall('.//a[@href]'): 

33 href = node.get('href') 

34 if href.find('/packages/') > -1: 

35 new_tarball = urlparse( 

36 href 

37 ) 

38 newhref = f"{self.config['pip']['self']}{new_tarball.path}" 

39 node.set('href', newhref) 

40 doc = ET.tostring(tree) 

41 

42 request['response'] = doc 

43 

44 @cherrypy.expose 

45 def proxy(self, environ, start_response): 

46 '''Proxy a pip repo request.''' 

47 path = environ["REQUEST_URI"].removeprefix(f"/{self.key}") 

48 

49 newrequest = {} 

50 newrequest['method'] = cherrypy.request.method 

51 newrequest['headers'] = cherrypy.request.headers 

52 newrequest['actual_request'] = cherrypy.request 

53 # application/vnd.pypi.simple.v1+json, application/vnd.pypi.simple.v1+html, and text/html 

54 

55 newrequest['content_type'] = 'application/vnd.pypi.simple.v1+html' 

56 newrequest['path'] = path 

57 

58 if path.startswith('/packages'): 

59 newrequest['storage'] = 'npm/tarballs' 

60 newhost = 'https://files.pythonhosted.org' 

61 self.logger.info( 

62 '%s Create new proxy with host %s and path %s', __name__, newhost, path) 

63 

64 dynamic_config = { 

65 'no_cache': self._proxy.no_cache, 

66 'logger': self.config['logger'], 

67 f"{self.key}": { 

68 'registry': newhost, 

69 } 

70 } 

71 dynamic_proxy = HttpProxy(dynamic_config, self.key) 

72 

73 return dynamic_proxy.rest_proxy(newrequest, start_response) 

74 

75 newrequest['storage'] = self.key 

76 newrequest['logger'] = self.logger 

77 newrequest['callback'] = self.callback 

78 return self._proxy.rest_proxy(newrequest, start_response)