openstack/glance
Code Issues Proposed changes

Merge "Fix S3 URL corruption on secret rotation"

Browse Source
This commit is contained in:
Zuul
2025-12-12 02:10:05 +00:00
committed by Gerrit Code Review
parent e8af40236f cbc4d9f5b1
commit ba82d3591d
3 changed files with 112 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

64
glance/common/store_utils.py
View File

@@ -239,18 +239,60 @@ def _update_cinder_location_and_store_id(context, loc):
return
def _update_s3_url(parsed, new_access_key, new_secret_key):
"""Update S3 URL with new credentials."""
host_part = parsed.netloc.split('@')[-1]
new_netloc = "%s:%s@%s" % (new_access_key, new_secret_key, host_part)
# Rebuild URL with new credentials but keep all other parts same
# We need to include params, query, fragment even if S3 URLs don't
# use them. This is to make sure we don't lose any URL parts
# when updating credentials
def _update_s3_url(url, new_access_key, new_secret_key):
"""Update S3 URL with new credentials.
This function replaces the access key and secret key in an S3 URL while
preserving all other URL components (host, port, path, query, fragment).
The implementation uses a hybrid approach:
- urlparse is used to extract query parameters and fragments, which it
handles correctly regardless of credential content.
- Manual string parsing with rfind('@') is used to locate credentials,
because urlparse incorrectly interprets '/' characters in the secret
key as path separators, corrupting the URL.
For example, with secret key 'abc/def':
- urlparse sees: s3://key:abc/def@host/bucket as path '/def@host/bucket'
- rfind('@') correctly finds the last '@' separating creds from host
"""
# Parse URL to get query and fragment (these are parsed correctly)
parsed = urlparse.urlparse(url)
scheme = parsed.scheme
# Find the last '@' in the original URL (before query/fragment)
# This handles the case where secret key contains '/'
url_without_query = url.split('?')[0].split('#')[0]
separator_position = url_without_query.rfind('@')
if separator_position == -1:
# No credentials - shouldn't happen for S3, but handle gracefully
new_netloc = "%s:%s@%s" % (new_access_key, new_secret_key,
parsed.netloc)
path_part = parsed.path # Use parsed.path when no '@' found
else:
# Extract host+path from after '@'
host_and_path = url_without_query[separator_position + 1:]
# Remove scheme part if it's still there
if '://' in host_and_path:
host_and_path = host_and_path.split('://', 1)[1]
# Separate host from path
first_slash = host_and_path.find('/')
if first_slash == -1:
host_part = host_and_path
path_part = ''
else:
host_part = host_and_path[:first_slash]
path_part = host_and_path[first_slash:]
new_netloc = "%s:%s@%s" % (new_access_key, new_secret_key, host_part)
# Rebuild URL preserving query and fragment
return urlparse.urlunparse((
parsed.scheme,
scheme,
new_netloc,
parsed.path,
path_part,
parsed.params,
parsed.query,
parsed.fragment
@@ -336,7 +378,7 @@ def _update_s3_location_and_store_id(context, loc):
new_access_key, new_secret_key = _get_store_credentials(
store_instance)
loc['url'] = _update_s3_url(
parsed, new_access_key, new_secret_key)
uri, new_access_key, new_secret_key)
return True
return False

2
glance/location.py
View File

@@ -134,7 +134,7 @@ def _update_location_for_legacy_store(image, image_repo):
LOG.debug("S3 Credentials mismatch, "
"updating URL")
loc['url'] = store_utils._update_s3_url(
parsed, new_access_key, new_secret_key)
loc['url'], new_access_key, new_secret_key)
# Save the image immediately after update
image_repo.save(image)
break

64
glance/tests/unit/common/test_utils.py
View File

@@ -1290,23 +1290,75 @@ class S3CredentialUpdateTestCase(test_utils.BaseTestCase):
def test_update_s3_url_helper_function(self):
"""Test the _update_s3_url helper function directly."""
# Test basic URL update
parsed = urlparse('s3://old_key:old_secret@bucket/object')
result = store_utils._update_s3_url(parsed, 'new_key', 'new_secret')
url = 's3://old_key:old_secret@bucket/object'
result = store_utils._update_s3_url(url, 'new_key', 'new_secret')
expected = 's3://new_key:new_secret@bucket/object'
self.assertEqual(result, expected)
# Test URL without existing credentials
parsed = urlparse('s3://bucket/object')
result = store_utils._update_s3_url(parsed, 'new_key', 'new_secret')
url = 's3://bucket/object'
result = store_utils._update_s3_url(url, 'new_key', 'new_secret')
expected = 's3://new_key:new_secret@bucket/object'
self.assertEqual(result, expected)
# Test with s3+https scheme
parsed = urlparse('s3+https://old_key:old_secret@bucket/object')
result = store_utils._update_s3_url(parsed, 'new_key', 'new_secret')
url = 's3+https://old_key:old_secret@bucket/object'
result = store_utils._update_s3_url(url, 'new_key', 'new_secret')
expected = 's3+https://new_key:new_secret@bucket/object'
self.assertEqual(result, expected)
def test_update_s3_url_with_slash_in_secret(self):
"""Test _update_s3_url handles forward slash in secret key.
AWS S3 secret access keys can contain '/'.
urlparse incorrectly interprets '/' as path separator, so we
use rfind('@') to correctly locate the credentials separator.
"""
# Secret key contains forward slash
url = 's3+https://old_key:old/secret@bucket/object'
result = store_utils._update_s3_url(url, 'new_key', 'new/secret')
expected = 's3+https://new_key:new/secret@bucket/object'
self.assertEqual(result, expected)
# Secret key contains multiple forward slashes
url = 's3+https://KEY:a/b/c@bucket/object'
result = store_utils._update_s3_url(url, 'NEW', 'x/y/z')
expected = 's3+https://NEW:x/y/z@bucket/object'
self.assertEqual(result, expected)
def test_update_s3_url_preserves_query_and_fragment(self):
"""Test _update_s3_url preserves query parameters and fragments."""
# URL with query parameters
url = 's3://old_key:old_secret@host.com/bucket/obj?param=value'
result = store_utils._update_s3_url(url, 'new_key', 'new_secret')
expected = 's3://new_key:new_secret@host.com/bucket/obj?param=value'
self.assertEqual(result, expected)
# URL with fragment
url = 's3://old_key:old_secret@host.com/bucket/obj#fragment'
result = store_utils._update_s3_url(url, 'new_key', 'new_secret')
expected = 's3://new_key:new_secret@host.com/bucket/obj#fragment'
self.assertEqual(result, expected)
# URL with both query and fragment
url = 's3://old_key:old_secret@host.com/bucket/obj?p=value#frag'
result = store_utils._update_s3_url(url, 'new_key', 'new_secret')
expected = 's3://new_key:new_secret@host.com/bucket/obj?p=value#frag'
self.assertEqual(result, expected)
def test_update_s3_url_with_port(self):
"""Test _update_s3_url handles hosts with port numbers."""
url = 's3://old_key:old_secret@host.com:8080/bucket/obj'
result = store_utils._update_s3_url(url, 'new_key', 'new_secret')
expected = 's3://new_key:new_secret@host.com:8080/bucket/obj'
self.assertEqual(result, expected)
# Port with slash in secret
url = 's3://old_key:old/secret@host.com:9000/bucket/obj'
result = store_utils._update_s3_url(url, 'new_key', 'new/secret')
expected = 's3://new_key:new/secret@host.com:9000/bucket/obj'
self.assertEqual(result, expected)
def test_construct_s3_url(self):
"""Test _construct_s3_url with all attributes present."""
store_instance = mock.Mock()