Test the CRLF fault injection using CORS

Signed-off-by: Tejas Chandramouli <tchandra@redhat.com>

rgw/v2/lib/curl/resource_op.py

@@ -10,7 +10,7 @@
 
 
 class CURL:
-    def __init__(self, user_info, ssh_con, ssl=None):
+    def __init__(self, user_info, ssh_con, curl_silent=True, ssl=None):
         """
         Constructor for curl class
         user_info(dict) : user details
@@ -19,7 +19,10 @@ def __init__(self, user_info, ssh_con, ssl=None):
         self.username = user_info["access_key"]
         self.password = user_info["secret_key"]
         self.endpoint_url = aws_reusable.get_endpoint(ssh_con, ssl)
-        self.prefix = f"curl --show-error --fail-with-body -v -s --aws-sigv4 aws:amz:us-east-1:s3 -u '{self.username}:{self.password}'"
+        if curl_silent:
+            self.prefix = f"curl --show-error --fail-with-body -v -s --aws-sigv4 aws:amz:us-east-1:s3 -u '{self.username}:{self.password}'"
+        else:
+            self.prefix = f"curl --show-error --fail-with-body -v --aws-sigv4 aws:amz:us-east-1:s3 -u '{self.username}:{self.password}'"
         if ssl:
             self.prefix = self.prefix + " --insecure"
 

rgw/v2/tests/curl/configs/test_crlf_injection_curl.yaml

@@ -0,0 +1,25 @@
+# script: test_cors_using_curl.py
+# polarion: CEPH-83574745
+config:
+  user_count: 1
+  bucket_count: 1
+  objects_count: 25
+  objects_size_range:
+    min: 5
+    max: 15
+  test_ops:
+    create_bucket: true
+    CRLF_injection: true
+    create_object: true
+    cors_origin: "http://www.cors-example.com"
+    policy_document:
+      "CORSRules":
+        [
+          {
+            "AllowedOrigins": ["http://www.cors-example.com"],
+            "AllowedHeaders": ["Authorization"],
+            "AllowedMethods": ["PUT", "GET"],
+            "ExposeHeaders": ["XXX\rArbitrary: Header\rAccess-Control-Allow-Credentials: true\rSet-Cookie: CRLF=Injection; \npath=/;"],
+            "MaxAgeSeconds": 3000,
+          },
+        ]

rgw/v2/tests/curl/reusable.py

@@ -268,11 +268,12 @@ def put_cors_object(
         input_file=s3_object_path,
         url_suffix=f"{bucket_name}/{s3_object_name}",
     )
-    upload_object_status = utils.exec_shell_cmd(command)
+    upload_object_status, err = utils.exec_shell_cmd(command, debug_info=True)
+    log.info(upload_object_status)
     if upload_object_status is False:
         raise TestExecError("object upload failed")
     log.info(f"object {s3_object_name} uploaded")
-    return True
+    return upload_object_status
 
 
 def download_object(

rgw/v2/tests/curl/test_cors_using_curl.py

@@ -1,10 +1,10 @@
 """
 Usage: test_cors_using_curl.py -c <input_yaml>
-polarion: CEPH-10355
+polarion: CEPH-10355, CEPH-83574745
 <input_yaml>
     Note: Following yaml can be used
     test_cors_using_curl.yaml
-    
+    test_crlf_injection_curl.yaml
 Operation:
 1. Create bucket and put CORS policy on the bucket.
 2. for the object count mentioned , upload objects using CURL calls with specified ORIGIN
@@ -58,7 +58,10 @@ def test_exec(config, ssh_con):
         cli_aws = AWS(ssl=config.ssl)
         endpoint = aws_reusable.get_endpoint(ssh_con, ssl=config.ssl)
         aws_auth.do_auth_aws(each_user)
-        curl_auth = CURL(each_user, ssh_con, ssl=config.ssl)
+        curl_silent = True
+        if config.test_ops.get("CRLF_injection", False):
+            curl_silent = False
+        curl_auth = CURL(each_user, ssh_con, curl_silent, ssl=config.ssl)
 
         for bc in range(config.bucket_count):
             bucket_name = utils.gen_bucket_name_from_userid(user_name, rand_no=bc)
@@ -76,6 +79,7 @@ def test_exec(config, ssh_con):
             log.info("Test CURL PUT,GET and DELETE with mentioned origin")
             # create objects
             objects_created_list = []
+
             # uploading data
             log.info(f"s3 objects to create: {config.objects_count}")
             for oc, size in list(config.mapped_sizes.items()):
@@ -84,29 +88,35 @@ def test_exec(config, ssh_con):
                 log.info(f"s3 object name: {s3_object_name}")
                 s3_object_path = os.path.join(TEST_DATA_PATH, s3_object_name)
                 log.info(f"s3 object path: {s3_object_path}")
-                curl_reusable.put_cors_object(
+                out = curl_reusable.put_cors_object(
                     curl_auth,
                     bucket_name,
                     s3_object_name,
                     TEST_DATA_PATH,
                     config,
                     cors_origin,
                 )
-                objects_created_list.append(s3_object_name)
-                curl_reusable.download_object(
-                    curl_auth,
-                    bucket_name,
-                    s3_object_name,
-                    TEST_DATA_PATH,
-                    s3_object_path,
-                    cors_origin,
-                )
-                curl_reusable.delete_object(
-                    curl_auth,
-                    bucket_name,
-                    s3_object_name,
-                    cors_origin,
-                )
+                if config.test_ops.get("CRLF_injection", False):
+                    if "\r" not in out:
+                        log.info("\r is sanitized as expected")
+                    else:
+                        raise TestExecError("\r is still present in curl output")
+                else:
+                    objects_created_list.append(s3_object_name)
+                    curl_reusable.download_object(
+                        curl_auth,
+                        bucket_name,
+                        s3_object_name,
+                        TEST_DATA_PATH,
+                        s3_object_path,
+                        cors_origin,
+                    )
+                    curl_reusable.delete_object(
+                        curl_auth,
+                        bucket_name,
+                        s3_object_name,
+                        cors_origin,
+                    )
             if config.local_file_delete is True:
                 log.info("deleting local file created after the upload")
                 utils.exec_shell_cmd(f"rm -rf  {s3_object_path}")