Coverage for core/data_backends.py: 37%

1import logging 

2import re 

3import tempfile 

4from contextlib import contextmanager 

5from pathlib import Path, PurePosixPath 

6from urllib.parse import urlparse 

7 

8from django import forms 

9from django.core.exceptions import ImproperlyConfigured 

10from django.utils.translation import gettext as _ 

11 

12from netbox.data_backends import DataBackend 

13from netbox.utils import register_data_backend 

14from utilities.constants import HTTP_PROXY_SUPPORTED_SCHEMAS, HTTP_PROXY_SUPPORTED_SOCK_SCHEMAS 

15from utilities.proxy import resolve_proxies 

16from utilities.socks import ProxyPoolManager 

17 

18from .exceptions import SyncError 

19 

20__all__ = ( 

21 'GitBackend', 

22 'LocalBackend', 

23 'S3Backend', 

24 'url_has_embedded_credentials', 

25) 

26 

27logger = logging.getLogger('netbox.data_backends') 

28 

29 

30def url_has_embedded_credentials(url): 

31 """ 

32 Check if a URL contains embedded credentials (username in the URL). 

33 

34 URLs like 'https://user@bitbucket.org/...' have embedded credentials. 

35 This is used to avoid passing explicit credentials to dulwich when the 

36 URL already contains them, which would cause authentication conflicts. 

37 """ 

38 parsed = urlparse(url) 

39 return bool(parsed.username) 

40 

41 

42@register_data_backend() 

43class LocalBackend(DataBackend): 

44 name = 'local' 

45 label = _('Local') 

46 is_local = True 

47 

48 @contextmanager 

49 def fetch(self): 

50 logger.debug("Data source type is local; skipping fetch") 

51 local_path = urlparse(self.url).path # Strip file:// scheme 

52 

53 yield local_path 

54 

55 

56@register_data_backend() 

57class GitBackend(DataBackend): 

58 name = 'git' 

59 label = 'Git' 

60 parameters = { 

61 'username': forms.CharField( 

62 required=False, 

63 label=_('Username'), 

64 widget=forms.TextInput(attrs={'class': 'form-control'}), 

65 help_text=_("Only used for cloning with HTTP(S)"), 

66 ), 

67 'password': forms.CharField( 

68 required=False, 

69 label=_('Password'), 

70 widget=forms.TextInput(attrs={'class': 'form-control'}), 

71 help_text=_("Only used for cloning with HTTP(S)"), 

72 ), 

73 'branch': forms.CharField( 

74 required=False, 

75 label=_('Branch'), 

76 widget=forms.TextInput(attrs={'class': 'form-control'}) 

77 ) 

78 } 

79 sensitive_parameters = ['password'] 

80 

81 def init_config(self): 

82 from dulwich.config import ConfigDict 

83 

84 # Initialize backend config 

85 config = ConfigDict() 

86 self.socks_proxy = None 

87 

88 # Apply HTTP proxy (if configured) 

89 proxies = resolve_proxies(url=self.url, context={'client': self}) or {} 

90 if proxy := proxies.get(self.url_scheme): 

91 if urlparse(proxy).scheme not in HTTP_PROXY_SUPPORTED_SCHEMAS: 

92 raise ImproperlyConfigured(f"Unsupported Git DataSource proxy scheme: {urlparse(proxy).scheme}") 

93 

94 if self.url_scheme in ('http', 'https'): 

95 config.set("http", "proxy", proxy) 

96 if urlparse(proxy).scheme in HTTP_PROXY_SUPPORTED_SOCK_SCHEMAS: 

97 self.socks_proxy = proxy 

98 

99 return config 

100 

101 @contextmanager 

102 def fetch(self): 

103 from dulwich import porcelain 

104 

105 local_path = tempfile.TemporaryDirectory() 

106 

107 clone_args = { 

108 "branch": self.params.get('branch'), 

109 "config": self.config, 

110 "errstream": porcelain.NoneStream(), 

111 } 

112 

113 # check if using socks for proxy - if so need to use custom pool_manager 

114 if self.socks_proxy: 

115 clone_args['pool_manager'] = ProxyPoolManager(self.socks_proxy) 

116 

117 if self.url_scheme in ('http', 'https'): 

118 # Only pass explicit credentials if URL doesn't already contain embedded username 

119 # to avoid credential conflicts (see #20902) 

120 if not url_has_embedded_credentials(self.url) and self.params.get('username'): 

121 clone_args.update( 

122 { 

123 "username": self.params.get('username'), 

124 "password": self.params.get('password'), 

125 } 

126 ) 

127 if self.url_scheme: 

128 clone_args["quiet"] = True 

129 clone_args["depth"] = 1 

130 

131 logger.debug(f"Cloning git repo: {self.url}") 

132 try: 

133 porcelain.clone(self.url, local_path.name, **clone_args) 

134 except BaseException as e: 

135 raise SyncError(_("Fetching remote data failed ({name}): {error}").format(name=type(e).__name__, error=e)) 

136 

137 yield local_path.name 

138 

139 local_path.cleanup() 

140 

141 

142@register_data_backend() 

143class S3Backend(DataBackend): 

144 name = 'amazon-s3' 

145 label = 'Amazon S3' 

146 parameters = { 

147 'aws_access_key_id': forms.CharField( 

148 label=_('AWS access key ID'), 

149 widget=forms.TextInput(attrs={'class': 'form-control'}) 

150 ), 

151 'aws_secret_access_key': forms.CharField( 

152 label=_('AWS secret access key'), 

153 widget=forms.TextInput(attrs={'class': 'form-control'}) 

154 ), 

155 } 

156 sensitive_parameters = ['aws_secret_access_key'] 

157 

158 REGION_REGEX = r's3\.([a-z0-9-]+)\.amazonaws\.com' 

159 

160 def init_config(self): 

161 from botocore.config import Config as Boto3Config 

162 

163 # Initialize backend config 

164 return Boto3Config( 

165 proxies=resolve_proxies(url=self.url, context={'client': self}), 

166 ) 

167 

168 @contextmanager 

169 def fetch(self): 

170 import boto3 

171 

172 local_path = tempfile.TemporaryDirectory() 

173 local_root = Path(local_path.name).resolve() 

174 

175 # Initialize the S3 resource and bucket 

176 aws_access_key_id = self.params.get('aws_access_key_id') 

177 aws_secret_access_key = self.params.get('aws_secret_access_key') 

178 s3 = boto3.resource( 

179 's3', 

180 region_name=self._region_name, 

181 aws_access_key_id=aws_access_key_id, 

182 aws_secret_access_key=aws_secret_access_key, 

183 config=self.config, 

184 endpoint_url=self._endpoint_url 

185 ) 

186 bucket = s3.Bucket(self._bucket_name) 

187 

188 try: 

189 # Download all files within the specified path 

190 for obj in bucket.objects.filter(Prefix=self._remote_path): 

191 local_filename = self._resolve_local_path(local_root, obj.key) 

192 # Build local path 

193 local_filename.parent.mkdir(parents=True, exist_ok=True) 

194 bucket.download_file(obj.key, str(local_filename)) 

195 

196 yield local_path.name 

197 finally: 

198 local_path.cleanup() 

199 

200 @staticmethod 

201 def _resolve_local_path(local_root, key): 

202 # S3 object keys are POSIX-style paths. Strip any leading separator so the key 

203 # joins onto the temp directory rather than replacing it, then ensure the 

204 # resolved destination remains within the temp directory to prevent path 

205 # traversal via crafted object keys. 

206 key_parts = PurePosixPath(key.lstrip('/')).parts 

207 local_filename = local_root.joinpath(*key_parts).resolve() 

208 if not local_filename.is_relative_to(local_root) or local_filename == local_root: 

209 raise SyncError( 

210 _("Invalid S3 object key '{key}': resolves outside of the local data directory").format(key=key) 

211 ) 

212 return local_filename 

213 

214 @property 

215 def _region_name(self): 

216 domain = urlparse(self.url).netloc 

217 if m := re.match(self.REGION_REGEX, domain): 

218 return m.group(1) 

219 return None 

220 

221 @property 

222 def _bucket_name(self): 

223 url_path = urlparse(self.url).path.lstrip('/') 

224 return url_path.split('/')[0] 

225 

226 @property 

227 def _endpoint_url(self): 

228 url_path = urlparse(self.url) 

229 return url_path._replace(params="", fragment="", query="", path="").geturl() 

230 

231 @property 

232 def _remote_path(self): 

233 url_path = urlparse(self.url).path.lstrip('/') 

234 if '/' in url_path: 

235 return url_path.split('/', 1)[1] 

236 return ''