Coverage for creepo/npmproxy.py: 69%

1"""An npm proxy""" 

2import io 

3import json 

4 

5from urllib.parse import urlparse 

6import urllib3 

7 

8import mime 

9 

10import cherrypy 

11 

12from httpproxy import HttpProxy 

13 

14 

15class NpmProxy: # pylint: disable=fixme 

16 """The npm proxy""" 

17 

18 def __init__(self, config): 

19 self.logger = config['logger'] 

20 self.config = config 

21 self.key = "npm" 

22 if self.key not in self.config: 

23 self.config[self.key] = {'registry': 'https://registry.npmjs.org'} 

24 

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

26 self.logger.debug('NpmProxy instantiated with %s', 

27 self.config[self.key]) 

28 

29 def callback(self, _input_bytes, request): 

30 """callback - preprocess the file before saving it""" 

31 

32 data = json.load(io.BytesIO(_input_bytes)) 

33 

34 for version in data['versions']: 

35 dist = data['versions'][version]['dist'] 

36 if dist: 

37 tarball = dist['tarball'] 

38 if tarball: 

39 # Point the client back to us for resolution. 

40 new_tarball = urllib3.util.Url( 

41 scheme='https', 

42 host='localhost', # TODO: get the listening public ip from config 

43 port=self.config['port'], 

44 path="/npm/tarballs/", 

45 query=tarball, 

46 ) 

47 data['versions'][version]['dist']['tarball'] = str( 

48 new_tarball) 

49 else: 

50 self.logger.warning( 

51 '%s Did not find tarball for %s', __name__, version) 

52 else: 

53 self.logger.warning( 

54 '%s Did not find dist for %s', __name__, version) 

55 request['response'] = bytes(json.dumps(data), 'utf-8') 

56 

57 @cherrypy.expose 

58 def proxy(self, environ, start_response): 

59 '''Proxy an npm request.''' 

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

61 newpath = path 

62 

63 newrequest = {} 

64 if len(mime.Types.of(path)) > 0: 

65 newrequest['content_type'] = mime.Types.of(path)[ 

66 0].content_type 

67 

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

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

70 newrequest['actual_request'] = cherrypy.request 

71 

72 if len(path.split('?')) == 2: 

73 new_remote = urlparse(path.split('?')[1]) 

74 newrequest['storage'] = f"{self.key}/tarballs" 

75 # The base proxy will remove the prefix 

76 newrequest['path'] = new_remote.path 

77 new_host = f"{new_remote.scheme}://{new_remote.netloc}" 

78 self.logger.info( 

79 '%s Create new proxy with host %s and path %s', __name__, new_host, new_remote.path) 

80 print( 

81 f"type(new_remote.path) is {type(new_remote.path)} {new_remote.path}") 

82 dynamic_proxy = HttpProxy( 

83 self._proxy.dynamic_config(new_host), self.key) 

84 

85 return dynamic_proxy.rest_proxy(newrequest, start_response) 

86 

87 newrequest['path'] = newpath 

88 newrequest['storage'] = 'npm' 

89 newrequest['content_type'] = 'application/octet-stream' 

90 self.logger.info('%s Requesting file %s', __name__, newpath) 

91 newrequest['logger'] = self._proxy.logger 

92 newrequest['callback'] = self.callback 

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