RDISCROWD-8392 upgrade to boto3

Author: peterkle

pybossa/cloud_store_api/connection.py

@@ -1,24 +1,15 @@
-from copy import deepcopy
-import ssl
-import sys
-import time
-
 from flask import current_app
-from boto.auth_handler import AuthHandler
-import boto.auth
-
-from boto.exception import S3ResponseError
-from boto.s3.key import Key
-from boto.s3.bucket import Bucket
-from boto.s3.connection import S3Connection, OrdinaryCallingFormat
-from boto.provider import Provider
-import jwt
+from botocore.config import Config
+import boto3
 from werkzeug.exceptions import BadRequest
 from boto3.session import Session
 from botocore.client import Config
 from pybossa.cloud_store_api.base_conn import BaseConnection
 from os import environ
 
+from pybossa.cloud_store_api.proxied_s3_client import ProxiedS3Client
+from pybossa.cloud_store_api.s3_client_wrapper import S3ClientWrapper
+
 
 def check_store(store):
     if not store:
@@ -60,59 +51,13 @@ def create_connection(**kwargs):
             proxy_url=kwargs.get("proxy_url")
         )
     if 'object_service' in kwargs:
-        current_app.logger.info("Calling ProxiedConnection")
-        conn = ProxiedConnection(**kwargs)
+        current_app.logger.info("Calling ProxiedS3Client")
+        conn = ProxiedS3Client(**kwargs)
     else:
-        current_app.logger.info("Calling CustomConnection")
-        conn = CustomConnection(**kwargs)
+        current_app.logger.info("Calling S3ClientWrapper")
+        conn = S3ClientWrapper(**kwargs)
     return conn
 
-
-class CustomProvider(Provider):
-    """Extend Provider to carry information about the end service provider, in
-       case the service is being proxied.
-    """
-
-    def __init__(self, name, access_key=None, secret_key=None,
-                 security_token=None, profile_name=None, object_service=None,
-                 auth_headers=None):
-        self.object_service = object_service or name
-        self.auth_headers = auth_headers
-        super(CustomProvider, self).__init__(name, access_key, secret_key,
-            security_token, profile_name)
-
-
-class CustomConnection(S3Connection):
-
-    def __init__(self, *args, **kwargs):
-        if not kwargs.get('calling_format'):
-            kwargs['calling_format'] = OrdinaryCallingFormat()
-
-        kwargs['provider'] = CustomProvider('aws',
-            kwargs.get('aws_access_key_id'),
-            kwargs.get('aws_secret_access_key'),
-            kwargs.get('security_token'),
-            kwargs.get('profile_name'),
-            kwargs.pop('object_service', None),
-            kwargs.pop('auth_headers', None))
-
-        kwargs['bucket_class'] = CustomBucket
-
-        ssl_no_verify = kwargs.pop('s3_ssl_no_verify', False)
-        self.host_suffix = kwargs.pop('host_suffix', '')
-
-        super(CustomConnection, self).__init__(*args, **kwargs)
-
-        if kwargs.get('is_secure', True) and ssl_no_verify:
-            self.https_validate_certificates = False
-            context = ssl._create_unverified_context()
-            self.http_connection_kwargs['context'] = context
-
-    def get_path(self, path='/', *args, **kwargs):
-        ret = super(CustomConnection, self).get_path(path, *args, **kwargs)
-        return self.host_suffix + ret
-
-
 class CustomConnectionV2(BaseConnection):
     def __init__(
         self,
@@ -133,89 +78,3 @@ def __init__(
                 proxies={"https": proxy_url, "http": proxy_url},
             ),
         )
-
-
-class CustomBucket(Bucket):
-    """Handle both 200 and 204 as response code"""
-
-    def delete_key(self, *args, **kwargs):
-        try:
-            super(CustomBucket, self).delete_key(*args, **kwargs)
-        except S3ResponseError as e:
-            if e.status != 200:
-                raise
-
-
-class ProxiedKey(Key):
-
-    def should_retry(self, response, chunked_transfer=False):
-        if 200 <= response.status <= 299:
-            return True
-        return super(ProxiedKey, self).should_retry(response, chunked_transfer)
-
-
-class ProxiedBucket(CustomBucket):
-
-    def __init__(self, *args, **kwargs):
-        super(ProxiedBucket, self).__init__(*args, **kwargs)
-        self.set_key_class(ProxiedKey)
-
-
-class ProxiedConnection(CustomConnection):
-    """Object Store connection through proxy API. Sets the proper headers and
-       creates the jwt; use the appropriate Bucket and Key classes.
-    """
-
-    def __init__(self, client_id, client_secret, object_service, *args, **kwargs):
-        self.client_id = client_id
-        self.client_secret = client_secret
-        kwargs['object_service'] = object_service
-        super(ProxiedConnection, self).__init__(*args, **kwargs)
-        self.set_bucket_class(ProxiedBucket)
-
-    def make_request(self, method, bucket='', key='', headers=None, data='',
-            query_args=None, sender=None, override_num_retries=None,
-            retry_handler=None):
-        headers = headers or {}
-        headers['jwt'] = self.create_jwt(method, self.host, bucket, key)
-        headers['x-objectservice-id'] = self.provider.object_service.upper()
-        current_app.logger.info("Calling ProxiedConnection.make_request. headers %s", str(headers))
-        return super(ProxiedConnection, self).make_request(method, bucket, key,
-            headers, data, query_args, sender, override_num_retries,
-            retry_handler)
-
-    def create_jwt(self, method, host, bucket, key):
-        now = int(time.time())
-        path = self.get_path(self.calling_format.build_path_base(bucket, key))
-        current_app.logger.info("create_jwt called. method %s, host %s, bucket %s, key %s, path %s", method, host, str(bucket), str(key), str(path))
-        payload = {
-            'iat': now,
-            'nbf': now,
-            'exp': now + 300,
-            'method': method,
-            'iss': self.client_id,
-            'host': host,
-            'path': path,
-            'region': 'ny'
-        }
-        return jwt.encode(payload, self.client_secret, algorithm='HS256')
-
-
-class CustomAuthHandler(AuthHandler):
-    """Implements sending of custom auth headers"""
-
-    capability = ['s3']
-
-    def __init__(self, host, config, provider):
-        if not provider.auth_headers:
-            raise boto.auth_handler.NotReadyToAuthenticate()
-        self._provider = provider
-        super(CustomAuthHandler, self).__init__(host, config, provider)
-
-    def add_auth(self, http_request, **kwargs):
-        headers = http_request.headers
-        for header, attr in self._provider.auth_headers:
-            headers[header] = getattr(self._provider, attr)
-
-    def sign_string(self, *args, **kwargs):
-        return ''

pybossa/cloud_store_api/proxied_s3_client.py

@@ -0,0 +1,171 @@
+from typing import Optional, Dict
+from urllib.parse import urlsplit, urlunsplit
+from botocore.exceptions import ClientError
+from botocore.config import Config
+import boto3
+import time
+import jwt
+import logging
+
+
+class ProxiedS3Client:
+    """
+    Emulates the old ProxiedConnection/ProxiedBucket/ProxiedKey behavior using boto3.
+
+    Features:
+      - Path-style addressing (OrdinaryCallingFormat equivalent)
+      - Optional SSL verification disable
+      - host_suffix is prepended to every request path (like get_path() override)
+      - Per-request JWT header and x-objectservice-id header
+      - Delete tolerant of HTTP 200 and 204
+    """
+
+    def __init__(
+        self,
+        *,
+        client_id: str,
+        client_secret: str,
+        object_service: str,
+        region_claim: str = "ny",               # value used in the JWT "region" claim
+        host_suffix: str = "",                  # prepended to every request path
+        extra_headers: Optional[Dict[str, str]] = None,      # any additional headers to inject
+        endpoint_url: Optional[str] = None,
+        region_name: Optional[str] = None,
+        profile_name: Optional[str] = None,
+        aws_access_key_id: Optional[str] = None,
+        aws_secret_access_key: Optional[str] = None,
+        aws_session_token: Optional[str] = None,
+        s3_ssl_no_verify: bool = False,
+        # optional logger with .info(...)
+        logger: Optional[logging.Logger] = None,
+    ):
+        self.client_id = client_id
+        self.client_secret = client_secret
+        self.object_service = object_service
+        self.region_claim = region_claim
+        self.host_suffix = host_suffix or ""
+        self.extra_headers = extra_headers or {}
+        self.logger = logger
+
+        session = (
+            boto3.session.Session(profile_name=profile_name)
+            if profile_name else boto3.session.Session()
+        )
+
+        config = Config(
+            region_name=region_name,
+            # OrdinaryCallingFormat equivalent
+            s3={"addressing_style": "path"},
+        )
+
+        verify = False if s3_ssl_no_verify else None  # None -> default verify
+
+        self.client = session.client(
+            "s3",
+            aws_access_key_id=aws_access_key_id,
+            aws_secret_access_key=aws_secret_access_key,
+            aws_session_token=aws_session_token,
+            endpoint_url=endpoint_url,
+            config=config,
+            verify=verify,
+        )
+
+        # One hook to: (1) prefix path, (2) add headers, (3) attach JWT
+        self.client.meta.events.register(
+            "before-sign.s3", self._before_sign_hook)
+
+    # ---------------------------
+    # Event hook: adjust request
+    # ---------------------------
+    def _before_sign_hook(self, request, operation_name, **kwargs):
+        """
+        request: botocore.awsrequest.AWSRequest
+        operation_name: e.g. "GetObject", "PutObject", etc.
+        """
+        parts = urlsplit(request.url)
+
+        # 1) Prefix request path with host_suffix (if set)
+        path = parts.path
+        if self.host_suffix:
+            path = (self.host_suffix.rstrip("/") + "/" +
+                    path.lstrip("/")).replace("//", "/")
+            request.url = urlunsplit(
+                (parts.scheme, parts.netloc, path, parts.query, parts.fragment))
+
+        # Recompute parts so host/path match the (possibly) updated URL
+        parts = urlsplit(request.url)
+        method = request.method
+        host = parts.netloc
+        path_for_jwt = parts.path  # include the prefixed path exactly as sent
+
+        # 2) Build headers (x-objectservice-id + any extra)
+        headers = dict(self.extra_headers)
+        headers["x-objectservice-id"] = self.object_service.upper()
+
+        # 3) Add JWT header
+        headers["jwt"] = self._create_jwt(method, host, path_for_jwt)
+
+        # Inject/override headers
+        for k, v in headers.items():
+            request.headers[k] = str(v)
+
+        if self.logger:
+            self.logger.info(
+                "ProxiedS3Client before-sign: op=%s method=%s host=%s path=%s headers=%s",
+                operation_name, method, host, path_for_jwt, list(
+                    headers.keys())
+            )
+
+    def _create_jwt(self, method: str, host: str, path: str) -> str:
+        now = int(time.time())
+        payload = {
+            "iat": now,
+            "nbf": now,
+            "exp": now + 300,         # 5 minutes
+            "method": method,
+            "iss": self.client_id,
+            "host": host,
+            "path": path,
+            "region": self.region_claim,
+        }
+        token = jwt.encode(payload, self.client_secret, algorithm="HS256")
+        # PyJWT may return bytes in older versions; ensure str
+        return token if isinstance(token, str) else token.decode("utf-8")
+
+    # ---------------------------
+    # Convenience helpers
+    # ---------------------------
+    def delete_key(self, bucket: str, key: str) -> bool:
+        """
+        Delete object: accept HTTP 200 or 204 as success (mirrors CustomBucket).
+        """
+        try:
+            resp = self.client.delete_object(Bucket=bucket, Key=key)
+            status = resp.get("ResponseMetadata", {}).get("HTTPStatusCode", 0)
+            if status not in (200, 204):
+                raise ClientError(
+                    {"Error": {"Code": str(status), "Message": "Unexpected status"},
+                     "ResponseMetadata": {"HTTPStatusCode": status}},
+                    operation_name="DeleteObject",
+                )
+            return True
+        except ClientError:
+            # Propagate non-success/delete errors
+            raise
+
+    def get_object(self, bucket: str, key: str, **kwargs):
+        return self.client.get_object(Bucket=bucket, Key=key, **kwargs)
+
+    def put_object(self, bucket: str, key: str, body, **kwargs):
+        return self.client.put_object(Bucket=bucket, Key=key, Body=body, **kwargs)
+
+    def list_objects(self, bucket: str, prefix: str = "", **kwargs):
+        return self.client.list_objects_v2(Bucket=bucket, Prefix=prefix, **kwargs)
+
+    def upload_file(self, filename: str, bucket: str, key: str, **kwargs):
+        # Uses s3transfer under the hood (built-in retries/backoff)
+        return self.client.upload_file(filename, bucket, key, ExtraArgs=kwargs or {})
+
+    def raw(self):
+        """Access the underlying boto3 client if you need operations not wrapped here."""
+        return self.client