wrap filenames in os.path.basename to prevent possible security exploit

add missing os import, fix a whitespace error

basename on esasky paths too

Author: keflavich

astroquery/alma/core.py

@@ -837,8 +837,8 @@ def download_files(self, files, *, savedir=None, cache=True,
                         raise (ex)
 
             try:
-                filename = re.search("filename=(.*)",
-                                     check_filename.headers['Content-Disposition']).groups()[0]
+                filename = os.path.basename(re.search("filename=(.*)",
+                                            check_filename.headers['Content-Disposition']).groups()[0])
             except KeyError:
                 log.info(f"Unable to find filename for {file_link}  "
                          "(missing Content-Disposition in header).  "

astroquery/esa/iso/core.py

@@ -12,6 +12,7 @@
 from astroquery.utils.tap.core import TapPlus
 from astroquery.query import BaseQuery
 import shutil
+import os
 from email.message import Message
 from requests import HTTPError
 from pathlib import Path
@@ -211,8 +212,8 @@ def get_postcard(self, tdt, *, filename=None, verbose=False):
             response = self._request('HEAD', link)
             response.raise_for_status()
 
-            filename = re.findall('filename="(.+)"', response.headers[
-                "Content-Disposition"])[0]
+            filename = os.path.basename(re.findall('filename="(.+)"', response.headers[
+                "Content-Disposition"])[0])
         else:
 
             filename = filename + ".png"

astroquery/esa/xmm_newton/core.py

@@ -182,8 +182,8 @@ def get_postcard(self, observation_id, *, image_type="OBS_EPIC",
         if filename is None:
             response = self._request('HEAD', link)
             response.raise_for_status()
-            filename = re.findall('filename="(.+)"', response.headers[
-                "Content-Disposition"])[0]
+            filename = os.path.basename(re.findall('filename="(.+)"', response.headers[
+                "Content-Disposition"])[0])
         else:
             filename = observation_id + ".png"
 
@@ -318,7 +318,7 @@ def _get_username_and_password(self, credentials_file):
 
     def _create_filename(self, filename, observation_id, suffixes):
         if filename is not None:
-            filename = os.path.splitext(filename)[0]
+            filename = os.path.basename(os.path.splitext(filename)[0])
         else:
             filename = observation_id
         filename += "".join(suffixes)

astroquery/esasky/core.py

@@ -1660,20 +1660,20 @@ def _extract_file_name_from_response_header(self, headers):
 
         if ".gz" in content_disposition[start_index:].lower():
             end_index = (content_disposition.lower().index(".gz", start_index + 1) + len(".gz"))
-            return content_disposition[start_index: end_index]
+            return os.path.basename(content_disposition[start_index: end_index])
         elif self.__FITS_STRING in content_disposition[start_index:].lower():
             end_index = (
                 content_disposition.lower().index(self.__FITS_STRING, start_index + 1) + len(self.__FITS_STRING))
-            return content_disposition[start_index: end_index]
+            return os.path.basename(content_disposition[start_index: end_index])
         elif self.__FTZ_STRING in content_disposition[start_index:].upper():
             end_index = (content_disposition.upper().index(self.__FTZ_STRING, start_index + 1) + len(self.__FTZ_STRING))
-            return content_disposition[start_index: end_index]
+            return os.path.basename(content_disposition[start_index: end_index])
         elif ".fit" in content_disposition[start_index:].upper():
             end_index = (content_disposition.upper().index(".fit", start_index + 1) + len(".fit"))
-            return content_disposition[start_index: end_index]
+            return os.path.basename(content_disposition[start_index: end_index])
         elif self.__TAR_STRING in content_disposition[start_index:].lower():
             end_index = (content_disposition.lower().index(self.__TAR_STRING, start_index + 1) + len(self.__TAR_STRING))
-            return content_disposition[start_index: end_index]
+            return os.path.basename(content_disposition[start_index: end_index])
         else:
             return ""
 

astroquery/utils/tap/conn/tapconn.py

@@ -19,7 +19,7 @@
 import mimetypes
 import platform
 import time
-
+import os
 from astroquery.utils.tap.xmlparser import utils
 from astroquery.utils.tap import taputils
 from astroquery import version
@@ -585,7 +585,7 @@ def get_file_from_header(self, headers):
         if content_disposition is not None:
             p = content_disposition.find('filename="')
             if p >= 0:
-                filename = content_disposition[p+10:len(content_disposition)-1]
+                filename = os.path.basename(content_disposition[p+10:len(content_disposition)-1])
                 content_encoding = self.find_header(headers, 'Content-Encoding')
                 if content_encoding is not None:
                     if "gzip" == content_encoding.lower():