Save credentials needed to download from S3 bucket to file in home directory. (6dbec3e4) · Commits · CAMS / CSO

src/cso/cso_dataspace.py

+89 −35

Original line number	Diff line number	Diff line
		@@ -49,6 +49,7 @@
		# 2026-03, Arjo Segers
		# Use STAC API to inquire Copernicus Dataspace.
		# Files are downloaded from S3 buckets.
		# Save credentials needed to download from S3 bucket to file in home directory.
		#


		@@ -625,10 +626,21 @@ class CSO_DataSpace_Downloader(object):
		"""

		# modules:
		import os
		import pathlib
		import requests

		# create?
		if (self.s3_credentials is None) or renew:
		# tools:
		from . import cso_file

		# local file with credentials:
		credfile = pathlib.Path.home() / ".cso" / "dataspace-credentials"

		# create new?
		if not os.path.isfile(credfile):

		# info ...
		logging.info(f"{indent}create S3 credentials ...")

		# fill authorization token in header:
		headers = {
		@@ -642,21 +654,18 @@ class CSO_DataSpace_Downloader(object):
		if r.status_code == 200:
		# evaluate:
		self.s3_credentials = r.json()
		# testing ...
		logging.info("{indent}Temporary S3 credentials created successfully.")
		logging.info(f" access: {self.s3_credentials['access_id']}")
		logging.info(f" secret: {self.s3_credentials['secret']}")
		# info ...
		logging.info(f"{indent} temporary S3 credentials created successfully.")
		## testing ..
		#logging.info(f"{indent} access: {self.s3_credentials['access_id']}")
		#logging.info(f"{indent} secret: {self.s3_credentials['secret']}")
		#
		# while testing, too many temporary credentials were create; re-use latest:
		# 403: Max number of credentials reached.
		elif r.status_code == 403:
		# info ...
		logging.warning(f"{indent}WARNING - re-using latest known credentials ...")
		# fill:
		self. s3_credentials = {
		"access_id" : "PANL68PPZGSERB1BO6TP",
		"secret" : "4f6e8llxSi9xKx4a4TX8EeJBPUIDUh8cwVaswGed",
		}
		logging.error(f"{indent} maximum number of credential exceeded, not deleted ...?")
		logging.error(f"{indent} try to delete manually on:")
		logging.error(f"{indent} https://eodata-s3keysmanager.dataspace.copernicus.eu/panel/s3-credentials")
		#
		else:
		logging.error(f"ERROR - failed to create temporary S3 credentials:")
		@@ -665,8 +674,52 @@ class CSO_DataSpace_Downloader(object):
		raise Exception
		#endif

		# info ...
		logging.info(f"{indent} store in {credfile} ...")
		# create target dir if necessary:
		cso_file.CheckDir( credfile )
		# write:
		with open(credfile,"w") as f:
		f.write( f"access_id = {self.s3_credentials['access_id']}\n" )
		f.write( f"secret = {self.s3_credentials['secret']}\n" )
		#endwith

		#endif # (re)new

		# read content:
		with open(credfile,"r") as f:
		lines = f.readlines()
		#endwith
		# target values:
		access_id = None
		secret = None
		# extract content:
		for line in lines:
		line = line.strip()
		if "=" in line:
		key,value = map( str.strip, line.split("=",1) )
		if key == "access_id":
		access_id = value
		elif key == "secret":
		secret = value
		#endif
		#endif
		#endif
		# check ..
		if access_id is None:
		logging.error(f"no 'access_id' found in credentials file: {credfile}")
		raise Exception
		#endif
		# check ..
		if secret is None:
		logging.error(f"no 'secret' found in credentials file: {secret}")
		raise Exception
		#endif
		# store:
		self.s3_credentials = {}
		self.s3_credentials['access_id'] = access_id
		self.s3_credentials['secret'] = secret

		#enddef Create_S3_Credentials

		# *
		@@ -799,6 +852,7 @@ class CSO_DataSpace_Downloader(object):
		# all ok, leave retry loop:
		break

		# unknown ...
		except Exception as err:
		logging.error(f"{indent}unknown error:")
		logging.error(f"{indent} {str(err)}")