crypto_bucket.py 18 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407
  1. # -*- coding: utf-8 -*-
  2. from . import http
  3. from . import exceptions
  4. from . import Bucket
  5. from .api import _make_range_string
  6. from .models import *
  7. from .compat import to_string, urlsplit, parse_qs
  8. from .crypto import BaseCryptoProvider
  9. from .exceptions import ClientError
  10. import copy
  11. import threading
  12. logger = logging.getLogger(__name__)
  13. class CryptoBucket(Bucket):
  14. """用于加密Bucket和Object操作的类,诸如上传、下载Object等。创建、删除bucket的操作需使用Bucket类接口。
  15. 用法(假设Bucket属于杭州区域) ::
  16. >>> import oss2
  17. >>> auth = oss2.Auth('your-access-key-id', 'your-access-key-secret')
  18. >>> bucket = oss2.CryptoBucket(auth, 'http://oss-cn-hangzhou.aliyuncs.com', 'your-bucket', oss2.LocalRsaProvider())
  19. >>> bucket.put_object('readme.txt', 'content of the object')
  20. <oss2.models.PutObjectResult object at 0x029B9930>
  21. :param auth: 包含了用户认证信息的Auth对象
  22. :type auth: oss2.Auth
  23. :param str endpoint: 访问域名或者CNAME
  24. :param str bucket_name: Bucket名
  25. :param crypto_provider: 客户端加密类。该参数默认为空
  26. :type crypto_provider: oss2.crypto.BaseCryptoProvider
  27. :param bool is_cname: 如果endpoint是CNAME则设为True;反之,则为False。
  28. :param session: 会话。如果是None表示新开会话,非None则复用传入的会话
  29. :type session: oss2.Session
  30. :param float connect_timeout: 连接超时时间,以秒为单位。
  31. :param str app_name: 应用名。该参数不为空,则在User Agent中加入其值。
  32. 注意到,最终这个字符串是要作为HTTP Header的值传输的,所以必须要遵循HTTP标准。
  33. :param bool enable_crc: 如果开启crc校验则设为True;反之,则为False
  34. """
  35. def __init__(self, auth, endpoint, bucket_name, crypto_provider,
  36. is_cname=False,
  37. session=None,
  38. connect_timeout=None,
  39. app_name='',
  40. enable_crc=True,
  41. proxies=None,
  42. region=None,
  43. cloudbox_id=None,
  44. is_path_style=False
  45. ):
  46. if not isinstance(crypto_provider, BaseCryptoProvider):
  47. raise ClientError('crypto_provider must be an instance of BaseCryptoProvider')
  48. logger.debug("Init CryptoBucket: {0}".format(bucket_name))
  49. super(CryptoBucket, self).__init__(auth, endpoint, bucket_name, is_cname, session, connect_timeout, app_name,
  50. enable_crc, proxies=proxies, region=region, cloudbox_id=cloudbox_id, is_path_style= is_path_style)
  51. self.crypto_provider = crypto_provider
  52. self.upload_contexts = {}
  53. self.upload_contexts_lock = threading.Lock()
  54. if self.app_name:
  55. self.user_agent = http.USER_AGENT + '/' + self.app_name + '/' + OSS_ENCRYPTION_CLIENT
  56. else:
  57. self.user_agent = http.USER_AGENT + '/' + OSS_ENCRYPTION_CLIENT
  58. def _init_user_agent(self, headers):
  59. if 'User-Agent' not in headers:
  60. headers['User-Agent'] = self.user_agent
  61. else:
  62. headers['User-Agent'] += '/' + OSS_ENCRYPTION_CLIENT
  63. def put_object(self, key, data,
  64. headers=None,
  65. progress_callback=None):
  66. """上传一个普通文件。
  67. 用法 ::
  68. >>> bucket.put_object('readme.txt', 'content of readme.txt')
  69. >>> with open(u'local_file.txt', 'rb') as f:
  70. >>> bucket.put_object('remote_file.txt', f)
  71. :param mat_desc: map,对象文件的description
  72. :param key: 上传到OSS的文件名
  73. :param data: 待上传的内容。
  74. :type data: bytes,str或file-like object
  75. :param headers: 用户指定的HTTP头部。可以指定Content-Type、Content-MD5、x-oss-meta-开头的头部等
  76. :type headers: 可以是dict,建议是oss2.CaseInsensitiveDict
  77. :param progress_callback: 用户指定的进度回调函数。可以用来实现进度条等功能。参考 :ref:`progress_callback` 。
  78. :return: :class:`PutObjectResult <oss2.models.PutObjectResult>`
  79. """
  80. logger.debug("Start to put object to CryptoBucket")
  81. headers = http.CaseInsensitiveDict(headers)
  82. self._init_user_agent(headers)
  83. content_crypto_material = self.crypto_provider.create_content_material()
  84. data = self.crypto_provider.make_encrypt_adapter(data, content_crypto_material.cipher)
  85. headers = content_crypto_material.to_object_meta(headers)
  86. return super(CryptoBucket, self).put_object(key, data, headers, progress_callback)
  87. def put_object_with_url(self, sign_url, data, headers=None, progress_callback=None):
  88. """ 使用加签的url上传对象
  89. :param sign_url: 加签的url
  90. :param data: 待上传的数据
  91. :param headers: 用户指定的HTTP头部。可以指定Content-Type、Content-MD5、x-oss-meta-开头的头部等,必须和签名时保持一致
  92. :param progress_callback: 用户指定的进度回调函数。参考 :ref:`progress_callback`
  93. :return:
  94. """
  95. raise ClientError("The operation is not support for CryptoBucket now")
  96. def append_object(self, key, position, data,
  97. headers=None,
  98. progress_callback=None,
  99. init_crc=None):
  100. raise ClientError("The operation is not support for CryptoBucket")
  101. def get_object(self, key,
  102. byte_range=None,
  103. headers=None,
  104. progress_callback=None,
  105. process=None,
  106. params=None):
  107. """下载一个文件。
  108. 用法 ::
  109. >>> result = bucket.get_object('readme.txt')
  110. >>> print(result.read())
  111. 'hello world'
  112. :param key: 文件名
  113. :param byte_range: 指定下载范围。参见 :ref:`byte_range`
  114. :param headers: HTTP头部
  115. :type headers: 可以是dict,建议是oss2.CaseInsensitiveDict
  116. :param progress_callback: 用户指定的进度回调函数。参考 :ref:`progress_callback`
  117. :param process: oss文件处理,如图像服务等。指定后process,返回的内容为处理后的文件。
  118. :param params: http 请求的查询字符串参数
  119. :type params: dict
  120. :return: file-like object
  121. :raises: 如果文件不存在,则抛出 :class:`NoSuchKey <oss2.exceptions.NoSuchKey>` ;还可能抛出其他异常
  122. """
  123. if process:
  124. raise ClientError("Process object operation is not support for Crypto Bucket")
  125. headers = http.CaseInsensitiveDict(headers)
  126. self._init_user_agent(headers)
  127. discard = 0
  128. range_string = ''
  129. if byte_range:
  130. if byte_range[0] is None and byte_range[1]:
  131. raise ClientError("Don't support range get while start is none and end is not")
  132. start, end = self.crypto_provider.adjust_range(byte_range[0], byte_range[1])
  133. adjust_byte_range = (start, end)
  134. range_string = _make_range_string(adjust_byte_range)
  135. if range_string:
  136. headers['range'] = range_string
  137. if byte_range[0] and adjust_byte_range[0] < byte_range[0]:
  138. discard = byte_range[0] - adjust_byte_range[0]
  139. logger.debug("adjust range of get object, byte_range: {0}, adjust_byte_range: {1}, discard: {2}".format(
  140. byte_range, adjust_byte_range, discard))
  141. logger.debug(
  142. "Start to get object from CryptoBucket: {0}, key: {1}, range: {2}, headers: {3}, params: {4}".format(
  143. self.bucket_name, to_string(key), range_string, headers, params))
  144. resp = self._do('GET', self.bucket_name, key, headers=headers, params=params)
  145. logger.debug("Get object done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
  146. return GetObjectResult(resp, progress_callback, self.enable_crc, crypto_provider=self.crypto_provider,
  147. discard=discard)
  148. def get_object_with_url(self, sign_url,
  149. byte_range=None,
  150. headers=None,
  151. progress_callback=None):
  152. """使用加签的url下载文件
  153. :param sign_url: 加签的url
  154. :param byte_range: 指定下载范围。参见 :ref:`byte_range`
  155. :param headers: HTTP头部
  156. :type headers: 可以是dict,建议是oss2.CaseInsensitiveDict,必须和签名时保持一致
  157. :param progress_callback: 用户指定的进度回调函数。参考 :ref:`progress_callback`
  158. :return: file-like object
  159. :raises: 如果文件不存在,则抛出 :class:`NoSuchKey <oss2.exceptions.NoSuchKey>` ;还可能抛出其他异常
  160. """
  161. query = parse_qs(urlsplit(sign_url).query)
  162. if query and (Bucket.PROCESS in query):
  163. raise ClientError("Process object operation is not support for Crypto Bucket")
  164. headers = http.CaseInsensitiveDict(headers)
  165. self._init_user_agent(headers)
  166. discard = 0
  167. range_string = ''
  168. if byte_range:
  169. if not byte_range[0] and byte_range[1]:
  170. raise ClientError("Don't support range get while start is none and end is not")
  171. start, end = self.crypto_provider.adjust_range(byte_range[0], byte_range[1])
  172. adjust_byte_range = (start, end)
  173. range_string = _make_range_string(adjust_byte_range)
  174. if range_string:
  175. headers['range'] = range_string
  176. if byte_range[0] and adjust_byte_range[0] < byte_range[0]:
  177. discard = byte_range[0] - adjust_byte_range[0]
  178. logger.debug("adjust range of get object, byte_range: {0}, adjust_byte_range: {1}, discard: {2}".format(
  179. byte_range, adjust_byte_range, discard))
  180. logger.debug(
  181. "Start to get object with url from CryptoBucket: {0}, sign_url: {1}, range: {2}, headers: {3}".format(
  182. self.bucket_name, sign_url, range_string, headers))
  183. resp = self._do_url('GET', sign_url, headers=headers)
  184. return GetObjectResult(resp, progress_callback, self.enable_crc,
  185. crypto_provider=self.crypto_provider, discard=discard)
  186. def create_select_object_meta(self, key, select_meta_params=None):
  187. raise ClientError("The operation is not support for Crypto Bucket")
  188. def select_object(self, key, sql,
  189. progress_callback=None,
  190. select_params=None
  191. ):
  192. raise ClientError("The operation is not support for CryptoBucket")
  193. def init_multipart_upload(self, key, headers=None, params=None, upload_context=None):
  194. """客户端加密初始化分片上传。
  195. :param params
  196. :param upload_context
  197. :param headers: HTTP头部
  198. :type headers: 可以是dict,建议是oss2.CaseInsensitiveDict
  199. :return: :class:`InitMultipartUploadResult <oss2.models.InitMultipartUploadResult>`
  200. 返回值中的 `crypto_multipart_context` 记录了加密Meta信息,在upload_part时需要一并传入
  201. """
  202. headers = http.CaseInsensitiveDict(headers)
  203. self._init_user_agent(headers)
  204. if not upload_context or not upload_context.data_size:
  205. raise ClientError("It is not support none upload_context and must specify data_size of upload_context ")
  206. logger.info("Start to init multipart upload by CryptoBucket, data_size: {0}, part_size: {1}".format(
  207. upload_context.data_size, upload_context.part_size))
  208. if upload_context.part_size:
  209. res = self.crypto_provider.cipher.is_valid_part_size(upload_context.part_size, upload_context.data_size)
  210. if not res:
  211. raise ClientError("part_size is invalid for multipart upload for CryptoBucket")
  212. else:
  213. upload_context.part_size = self.crypto_provider.cipher.determine_part_size(upload_context.data_size)
  214. content_crypto_material = self.crypto_provider.create_content_material()
  215. upload_context.content_crypto_material = content_crypto_material
  216. headers = content_crypto_material.to_object_meta(headers, upload_context)
  217. resp = super(CryptoBucket, self).init_multipart_upload(key, headers)
  218. return resp
  219. def upload_part(self, key, upload_id, part_number, data, progress_callback=None, headers=None, upload_context=None):
  220. """客户端加密上传一个分片。
  221. :param upload_context:
  222. :param str key: 待上传文件名,这个文件名要和 :func:`init_multipart_upload` 的文件名一致。
  223. :param str upload_id: 分片上传ID
  224. :param int part_number: 分片号,最小值是1.
  225. :param data: 待上传数据。
  226. :param progress_callback: 用户指定进度回调函数。可以用来实现进度条等功能。参考 :ref:`progress_callback` 。
  227. :param headers: 用户指定的HTTP头部。可以指定Content-MD5头部等
  228. :type headers: 可以是dict,建议是oss2.CaseInsensitiveDict
  229. :return: :class:`PutObjectResult <oss2.models.PutObjectResult>`
  230. """
  231. logger.info(
  232. "Start to upload multipart of CryptoBucket, upload_id = {0}, part_number = {1}".format(upload_id,
  233. part_number))
  234. headers = http.CaseInsensitiveDict(headers)
  235. self._init_user_agent(headers)
  236. if upload_context:
  237. context = upload_context
  238. else:
  239. raise ClientError("Could not init upload context, upload contexts flag is False and upload context is none")
  240. content_crypto_material = context.content_crypto_material
  241. if content_crypto_material.cek_alg != self.crypto_provider.cipher.alg or content_crypto_material.wrap_alg != \
  242. self.crypto_provider.wrap_alg:
  243. err_msg = 'Envelope or data encryption/decryption algorithm is inconsistent'
  244. raise InconsistentError(err_msg, self)
  245. headers = content_crypto_material.to_object_meta(headers, context)
  246. plain_key = self.crypto_provider.decrypt_encrypted_key(content_crypto_material.encrypted_key)
  247. plain_iv = self.crypto_provider.decrypt_encrypted_iv(content_crypto_material.encrypted_iv)
  248. offset = context.part_size * (part_number - 1)
  249. counter = self.crypto_provider.cipher.calc_offset(offset)
  250. cipher = copy.copy(content_crypto_material.cipher)
  251. cipher.initialize(plain_key, plain_iv, counter)
  252. data = self.crypto_provider.make_encrypt_adapter(data, cipher)
  253. resp = super(CryptoBucket, self).upload_part(key, upload_id, part_number, data, progress_callback, headers)
  254. return resp
  255. def complete_multipart_upload(self, key, upload_id, parts, headers=None):
  256. """客户端加密完成分片上传,创建文件。
  257. 当所有分片均已上传成功,才可以调用此函数
  258. :param str key: 待上传的文件名,这个文件名要和 :func:`init_multipart_upload` 的文件名一致。
  259. :param str upload_id: 分片上传ID
  260. :param parts: PartInfo列表。PartInfo中的part_number和etag是必填项。其中的etag可以从 :func:`upload_part` 的返回值中得到。
  261. :type parts: list of `PartInfo <oss2.models.PartInfo>`
  262. :param headers: HTTP头部
  263. :type headers: 可以是dict,建议是oss2.CaseInsensitiveDict
  264. :return: :class:`PutObjectResult <oss2.models.PutObjectResult>`
  265. """
  266. logger.info("Start to complete multipart upload of CryptoBucket, upload_id = {0}".format(upload_id))
  267. headers = http.CaseInsensitiveDict(headers)
  268. self._init_user_agent(headers)
  269. try:
  270. resp = super(CryptoBucket, self).complete_multipart_upload(key, upload_id, parts, headers)
  271. except exceptions as e:
  272. raise e
  273. return resp
  274. def abort_multipart_upload(self, key, upload_id, headers=None):
  275. """取消分片上传。
  276. :param headers: 可以是dict,建议是oss2.CaseInsensitiveDict
  277. :param str key: 待上传的文件名,这个文件名要和 :func:`init_multipart_upload` 的文件名一致。
  278. :param str upload_id: 分片上传ID
  279. :return: :class:`RequestResult <oss2.models.RequestResult>`
  280. """
  281. logger.info("Start to abort multipart upload of CryptoBucket, upload_id = {0}".format(upload_id))
  282. headers = http.CaseInsensitiveDict(headers)
  283. self._init_user_agent(headers)
  284. try:
  285. resp = super(CryptoBucket, self).abort_multipart_upload(key, upload_id)
  286. except exceptions as e:
  287. raise e
  288. return resp
  289. def upload_part_copy(self, source_bucket_name, source_key, byte_range, target_key, target_upload_id,
  290. target_part_number, headers=None):
  291. """分片拷贝。把一个已有文件的一部分或整体拷贝成目标文件的一个分片。
  292. :param target_part_number:
  293. :param target_upload_id:
  294. :param target_key:
  295. :param source_key:
  296. :param source_bucket_name:
  297. :param byte_range: 指定待拷贝内容在源文件里的范围。参见 :ref:`byte_range`
  298. :param headers: HTTP头部
  299. :type headers: 可以是dict,建议是oss2.CaseInsensitiveDict
  300. :return: :class:`PutObjectResult <oss2.models.PutObjectResult>`
  301. """
  302. raise ClientError("The operation is not support for CryptoBucket now")
  303. def process_object(self, key, process):
  304. raise ClientError("The operation is not support for CryptoBucket")