Repository: RaRe-Technologies/smart_open
Pull request #776 (2023-07-19 02:08:34)

Improve S3 URI Parsing for URIs with "@", "/", and ":"
#### Improve S3 URI Parsing for URIs with "@", "/", and ":"

#### Motivation

I am improving an old PR of mine from a long time ago and addressing a recent comment:
https://github.com/RaRe-Technologies/smart_open/pull/235
https://github.com/RaRe-Technologies/smart_open/pull/235#issuecomment-1640707965

Original Issue: https://github.com/RaRe-Technologies/smart_open/issues/234

#### Tests

I wrote two unit tests which pass and no regressions.

#### Checklist

Before you create the PR, please make sure you have:

- [x] Picked a concise, informative and complete title
- [x] Clearly explained the motivation behind the PR
- [x] Linked to any existing issues that your PR will be solving
- [x] Included tests for any new functionality
- [x] Checked that all unit tests pass



Patch:
diff --git a/smart_open/s3.py b/smart_open/s3.py
index b211163..d8ce60c 100644
--- a/smart_open/s3.py
+++ b/smart_open/s3.py
@@ -105,9 +105,20 @@ def parse_uri(uri_as_string):
     #
     uri = split_uri.netloc + split_uri.path
 
-    if '@' in uri and ':' in uri.split('@')[0]:
-        auth, uri = uri.split('@', 1)
-        access_id, access_secret = auth.split(':')
+    #
+    # Attempt to extract edge-case authentication details from the URL.
+    #
+    # See:
+    #   1. https://summitroute.com/blog/2018/06/20/aws_security_credential_formats/
+    #   2. test_s3_uri_with_credentials* in test_smart_open.py for example edge cases
+    #
+    if '@' in uri:
+        maybe_auth, rest = uri.split('@', 1)
+        if ':' in maybe_auth:
+            maybe_id, maybe_secret = maybe_auth.split(':', 1)
+            if '/' not in maybe_id:
+                access_id, access_secret = maybe_id, maybe_secret
+                uri = rest
 
     head, key_id = uri.split('/', 1)
     if '@' in head and ':' in head:


Test patch:
diff --git a/smart_open/tests/test_smart_open.py b/smart_open/tests/test_smart_open.py
index 367e994..33a0cf8 100644
--- a/smart_open/tests/test_smart_open.py
+++ b/smart_open/tests/test_smart_open.py
@@ -132,6 +132,17 @@ class ParseUriTest(unittest.TestCase):
         self.assertEqual(parsed_uri.access_id, "accessid")
         self.assertEqual(parsed_uri.access_secret, "access/secret")
 
+    #
+    # Nb. should never happen in theory, but if it does, we should avoid crashing
+    #
+    def test_s3_uri_has_colon_in_secret(self):
+        parsed_uri = smart_open_lib._parse_uri("s3://accessid:access/secret:totally@mybucket/my@ke@y")
+        self.assertEqual(parsed_uri.scheme, "s3")
+        self.assertEqual(parsed_uri.bucket_id, "mybucket")
+        self.assertEqual(parsed_uri.key_id, "my@ke@y")
+        self.assertEqual(parsed_uri.access_id, "accessid")
+        self.assertEqual(parsed_uri.access_secret, "access/secret:totally")
+
     def test_s3_uri_has_atmark_in_key_name2(self):
         parsed_uri = smart_open_lib._parse_uri(
             "s3://accessid:access/secret@hostname:1234@mybucket/dir/my@ke@y"
@@ -218,6 +229,24 @@ class ParseUriTest(unittest.TestCase):
         self.assertEqual(parsed_uri.access_id, None)
         self.assertEqual(parsed_uri.access_secret, None)
 
+    def test_s3_uri_with_at_symbol_in_key_name0(self):
+        """ Correctly parse the s3 url if there is an @ symbol (and colon) in the key or dir """
+        parsed_uri = smart_open_lib._parse_uri("s3://mybucket/mydir:my@key")
+        self.assertEqual(parsed_uri.scheme, "s3")
+        self.assertEqual(parsed_uri.bucket_id, "mybucket")
+        self.assertEqual(parsed_uri.key_id, "mydir:my@key")
+        self.assertEqual(parsed_uri.access_id, None)
+        self.assertEqual(parsed_uri.access_secret, None)
+
+    def test_s3_uri_with_at_symbol_in_key_name1(self):
+        """ Correctly parse the s3 url if there is an @ symbol (and colon) in the key or dir """
+        parsed_uri = smart_open_lib._parse_uri("s3://mybucket/my:dir@my/key")
+        self.assertEqual(parsed_uri.scheme, "s3")
+        self.assertEqual(parsed_uri.bucket_id, "mybucket")
+        self.assertEqual(parsed_uri.key_id, "my:dir@my/key")
+        self.assertEqual(parsed_uri.access_id, None)
+        self.assertEqual(parsed_uri.access_secret, None)
+
     def test_s3_uri_contains_question_mark(self):
         parsed_uri = smart_open_lib._parse_uri("s3://mybucket/mydir/mykey?param")
         self.assertEqual(parsed_uri.scheme, "s3")