Add SignedPolicyDocument support to storage.py and fast_http.py

src/clusterfuzz/_internal/google_cloud_utils/storage.py

@@ -118,6 +118,22 @@
                                                  ['url', 'filepath'])
 
 
+class SignedPolicyDocument:
+  """Signed policy document."""
+
+  def __init__(self, bucket, policy, x_goog_algorithm, x_goog_date,
+               x_goog_credential, x_goog_signature):
+    self.bucket = bucket
+    self.policy = policy
+    self.x_goog_algorithm = x_goog_algorithm
+    self.x_goog_date = x_goog_date
+    self.x_goog_credential = x_goog_credential
+    self.x_goog_signature = x_goog_signature
+
+  def to_json(self):
+    return json.dumps(self.__dict__)
+
+
 class StorageProvider:
   """Core storage provider interface."""
 
@@ -1314,6 +1330,36 @@ def get_signed_upload_url(remote_path, minutes=SIGNED_URL_EXPIRATION_MINUTES):
   return provider.sign_upload_url(remote_path, minutes=minutes)
 
 
+def get_signed_policy_document(remote_path,
+                               minutes=SIGNED_URL_EXPIRATION_MINUTES):
+  """Returns a signed policy document for |remote_path|."""
+  if _integration_test_env_doesnt_support_signed_urls():
+    return None
+
+  minutes = datetime.timedelta(minutes=minutes)
+  bucket_name, object_path = get_bucket_name_and_path(remote_path)
+  signing_creds, access_token = _signing_creds()
+  client = _storage_client()
+  bucket = client.bucket(bucket_name)
+  conditions = [['starts-with', '$key', object_path]]
+  policy = bucket.generate_signed_post_policy_v4(
+      object_path,
+      expiration=minutes,
+      conditions=conditions,
+      credentials=signing_creds,
+      access_token=access_token,
+      service_account_email=signing_creds.service_account_email)
+
+  fields = policy['fields']
+  return SignedPolicyDocument(
+      bucket=bucket_name,
+      policy=fields['policy'],
+      x_goog_algorithm=fields['x-goog-algorithm'],
+      x_goog_date=fields['x-goog-date'],
+      x_goog_credential=fields['x-goog-credential'],
+      x_goog_signature=fields['x-goog-signature'])
+
+
 def get_signed_download_url(remote_path, minutes=SIGNED_URL_EXPIRATION_MINUTES):
   """Returns a signed download URL for |remote_path|. Does not download the
   contents."""

src/clusterfuzz/_internal/system/fast_http.py

@@ -14,6 +14,7 @@
 """Tools for fast HTTP operations."""
 import asyncio
 import itertools
+import os
 from typing import List
 from typing import Sequence
 from typing import Tuple
@@ -92,6 +93,24 @@ async def _async_download_file(session: aiohttp.ClientSession, url: str,
         file_handle.write(chunk)
 
 
+async def upload_using_signed_policy_document(local_path, upload_path, doc):
+  """Uploads a file using a signed policy document."""
+  url = f'https://storage.googleapis.com/{doc.bucket}'
+  data = aiohttp.FormData()
+  data.add_field('key', upload_path)
+  data.add_field('policy', doc.policy)
+  data.add_field('x-goog-algorithm', doc.x_goog_algorithm)
+  data.add_field('x-goog-date', doc.x_goog_date)
+  data.add_field('x-goog-credential', doc.x_goog_credential)
+  data.add_field('x-goog-signature', doc.x_goog_signature)
+
+  with open(local_path, 'rb') as f:
+    data.add_field('file', f, filename=os.path.basename(local_path))
+    async with aiohttp.ClientSession(timeout=_HTTP_TIMEOUT_SECONDS) as session:
+      async with session.post(url, data=data) as response:
+        response.raise_for_status()
+
+
 def _get_blob_url(bucket_name, blob_name):
   blob_name = urllib.parse.quote(blob_name, safe='')
   return (