support path-scoped registry credentials

projects/path-scoped-registry-credentials.md

@@ -0,0 +1,116 @@
+<!--
+SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION. All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+
+SPDX-License-Identifier: Apache-2.0
+-->
+
+# Path-Scoped Registry Credentials
+
+**Author**: [Trasha Dewan](https://github.com/tdewanNvidia)<br>
+**PIC**: [Trasha Dewan](https://github.com/tdewanNvidia)<br>
+**Proposal Issue**: [#1113](https://github.com/nvidia/osmo/issues/1113)
+
+## Overview
+
+OSMO should allow registry credentials to be scoped to a registry path, not only
+a registry host. This lets workflows select the correct credential when multiple
+repositories share the same registry host.
+
+### Motivation
+
+Docker and Kubernetes image pull configuration supports matching credentials by
+registry path. OSMO should preserve that behavior so users can safely configure
+separate credentials for separate repository namespaces under one registry host.
+
+### Problem
+
+OSMO previously normalized registry credentials to host-only keys. That made
+path-specific credentials indistinguishable and could cause workflow validation
+or generated image pull secrets to use the wrong credential for an image path.
+
+## Use Cases
+
+| Use Case | Description |
+|---|---|
+| Separate credentials by path | A user configures different credentials for different repository paths under the same registry host. |
+| Pull workflow images | OSMO validates and pulls an image using only credentials whose scope matches that image path. |
+| Preserve host-level fallback | A host-level credential still applies when no more specific path-scoped credential exists. |
+
+## Requirements
+
+| Title | Description | Type |
+|---|---|---|
+| Preserve registry path | OSMO shall store the full registry path supplied for a registry credential. | Functional |
+| Path-aware matching | OSMO shall match image references to credentials by registry host and path segment. | Functional |
+| Specificity ordering | OSMO shall try more specific matching scopes before less specific scopes. | Functional |
+| Pull secret generation | OSMO shall include matching registry credentials in generated image pull secrets. | Functional |
+| Backwards compatibility | Existing host-scoped registry credentials shall continue to work. | Functional |
+
+## Architectural Details
+
+The change adds shared registry-scope helpers and routes all registry credential
+lookup paths through them. A registry scope is the registry host plus an optional
+repository path. Matching is segment-aware, so a scope for one sibling path does
+not match another sibling path.
+
+Key components:
+
+| Component | Purpose |
+|---|---|
+| `src/lib/utils/common.py` | Normalizes registry scopes and provides matching helpers. |
+| `src/utils/connectors/postgres.py` | Returns registry credentials keyed by normalized scope and decrypts only matching credentials. |
+| `src/utils/job/workflow.py` | Validates private workflow images against matching user credentials. |
+| `src/utils/job/task.py` | Builds image pull secrets from matching registry credentials. |
+| `src/service/core/workflow/objects.py` | Validates registry credential creation using the normalized scope. |
+
+## Detailed Design
+
+Registry profiles are normalized into a canonical scope:
+
+```text
+<registry-host>[/repository/path]
+```
+
+Default HTTPS port notation is canonicalized so equivalent scopes compare the
+same way. Non-default ports remain part of the scope.
+
+When validating or preparing a workflow image, OSMO:
+
+1. Parses the image reference.
+2. Computes the image registry scope.
+3. Finds all stored credential scopes that contain the image scope.
+4. Orders matches from most specific to least specific.
+5. Uses only those matching credentials for validation and image pull secret
+   generation.
+
+## Backwards Compatibility
+
+Existing host-scoped credentials continue to match all image paths under the
+same registry host. Path-scoped credentials add a more specific option without
+changing the host-level behavior.
+
+## Security
+
+Credential matching is narrowed to the image path being used. The filtered
+lookup avoids decrypting unrelated registry credentials during image validation.
+
+## Testing
+
+- Unit tests cover scope normalization, matching, workflow validation, and pull secret generation.
+- Manual CLI validation confirms distinct path-scoped credentials are stored and matched separately.
+
+## Open Questions
+
+- None.

src/cli/credential.py

@@ -198,7 +198,7 @@ def setup_parser(parser: argparse._SubParsersAction):
     set_parser = subparsers.add_parser('set', help='Create or update a credential',
         formatter_class=argparse.RawTextHelpFormatter,
         epilog='Ex. osmo credential set registry_cred_name --type REGISTRY ' +
-        '--payload registry=your_registry username=your_username auth=xxxxxx \n' +
+        '--payload registry=your_registry_or_path username=your_username auth=xxxxxx \n' +
         'Ex. osmo credential set data_cred_name --type DATA ' +
         '--payload access_key_id=your_s3_username access_key=xxxxxx ' +
         'endpoint=s3://bucket \n' +

src/lib/utils/common.py

@@ -30,6 +30,7 @@
 import threading
 import time
 from typing import Annotated, Any, Callable, Coroutine, Dict, Generator, Iterable, Iterator, List, NamedTuple, Optional, Set, Tuple
+from urllib.parse import urlparse
 import uuid
 
 import pydantic
@@ -449,6 +450,60 @@ def registry_parse(name: str) -> str:
     return name
 
 
+def normalize_registry_scope(registry_scope: str) -> str:
+    """Normalize a registry credential scope while preserving path prefixes."""
+    scope = registry_scope.strip().rstrip('/')
+    if not scope:
+        return DEFAULT_REGISTRY
+
+    parsed = urlparse(scope if '://' in scope else f'//{scope}')
+    if parsed.netloc:
+        registry = registry_parse(parsed.netloc)
+        if registry.endswith(':443'):
+            registry = registry[:-4]
+        path = parsed.path.strip('/')
+        if path:
+            return f'{registry}/{path}'
+        return registry
+
+    return registry_parse(scope)
+
+
+def image_registry_scope(image_info: DockerImageInfo) -> str:
+    """Return the registry scope string Kubernetes matches credentials against."""
+    registry = image_info.host
+    if image_info.port != 443:
+        registry = f'{registry}:{image_info.port}'
+    return f'{registry}/{image_info.name}'
+
+
+def registry_scope_contains(parent_scope: str, child_scope: str) -> bool:
+    """Return whether child_scope is inside parent_scope by path segment."""
+    parent = normalize_registry_scope(parent_scope)
+    child = normalize_registry_scope(child_scope)
+    return child == parent or child.startswith(f'{parent}/')
+
+
+def registry_scope_matches_image(registry_scope: str, image_info: DockerImageInfo) -> bool:
+    """Return whether a registry credential scope applies to an image."""
+    return registry_scope_contains(registry_scope, image_registry_scope(image_info))
+
+
+def matching_registry_scopes(
+    image_info: DockerImageInfo,
+    registry_scopes: Iterable[str],
+) -> List[str]:
+    """Return all credential scopes that match an image, most specific first."""
+    matches = [
+        registry_scope for registry_scope in registry_scopes
+        if registry_scope_matches_image(registry_scope, image_info)
+    ]
+    return sorted(
+        matches,
+        key=lambda registry_scope: len(normalize_registry_scope(registry_scope)),
+        reverse=True)
+
+
 def docker_parse(image: str) -> DockerImageInfo:
     """ Parses a docker image into its separate components """
     # Parse image according to rules here

src/lib/utils/tests/test_common.py

@@ -104,6 +104,49 @@ def test_docker_parse(self):
                 self.assertEqual(result.tag, exp_tag, f'tag mismatch for {image}')
                 self.assertEqual(result.digest, exp_digest, f'digest mismatch for {image}')
 
+    def test_normalize_registry_scope_preserves_path(self):
+        self.assertEqual(common.normalize_registry_scope('docker.io'), common.DEFAULT_REGISTRY)
+        self.assertEqual(
+            common.normalize_registry_scope('https://nvcr.io/nvstaging/osmo/'),
+            'nvcr.io/nvstaging/osmo')
+        self.assertEqual(
+            common.normalize_registry_scope('localhost:5000/team/image'),
+            'localhost:5000/team/image')
+        self.assertEqual(
+            common.normalize_registry_scope('registry.example.com:443/team/image'),
+            'registry.example.com/team/image')
+
+    def test_registry_scope_matches_image_by_path_segment(self):
+        image_info = common.docker_parse('nvcr.io/nvstaging/osmo/app:latest')
+
+        self.assertTrue(common.registry_scope_matches_image('nvcr.io', image_info))
+        self.assertTrue(
+            common.registry_scope_matches_image('nvcr.io/nvstaging/osmo', image_info))
+        self.assertFalse(
+            common.registry_scope_matches_image('nvcr.io/nvstaging/isaac', image_info))
+        self.assertFalse(
+            common.registry_scope_matches_image('nvcr.io/nvstaging/osmosis', image_info))
+
+    def test_matching_registry_scopes_returns_most_specific_first(self):
+        image_info = common.docker_parse('nvcr.io/nvstaging/osmo/app:latest')
+
+        matches = common.matching_registry_scopes(image_info, [
+            'nvcr.io',
+            'nvcr.io/nvstaging/isaac',
+            'nvcr.io/nvstaging/osmo',
+        ])
+
+        self.assertEqual(matches, ['nvcr.io/nvstaging/osmo', 'nvcr.io'])
+
+    def test_image_registry_scope_includes_non_default_port(self):
+        image_info = common.docker_parse('registry.example.com:5000/org/image:v1')
+
+        self.assertEqual(
+            common.image_registry_scope(image_info),
+            'registry.example.com:5000/org/image')
+        self.assertTrue(
+            common.registry_scope_matches_image('registry.example.com:5000/org', image_info))
+
 
 class TestPydanticEncoder(unittest.TestCase):
     """ Tests for pydantic_encoder. """

src/service/core/workflow/objects.py

@@ -631,10 +631,14 @@ def to_db_row(self, user: str, postgres: connectors.PostgresConnector) -> Creden
                                 connectors.PostgresConnector.encode_hstore(payload))
 
     def valid_cred(self, workflow_config: connectors.WorkflowConfig):
-        self.registry = common.registry_parse(self.registry)
-        if self.registry in workflow_config.credential_config.disable_registry_validation:
+        self.registry = common.normalize_registry_scope(self.registry)
+        if any(
+            common.registry_scope_contains(disabled_registry, self.registry)
+            for disabled_registry in workflow_config.credential_config.disable_registry_validation
+        ):
             return
-        response = common.registry_auth(f'https://{self.registry}/v2/',
+        registry_host = self.registry.split('/', 1)[0]
+        response = common.registry_auth(f'https://{registry_host}/v2/',
                                         self.username, self.auth)
         if response.status_code != 200:
             raise osmo_errors.OSMOCredentialError('Registry authentication failed.')

src/service/core/workflow/tests/test_helpers.py

@@ -77,6 +77,26 @@ def make_workflow(name: str, groups: list) -> objects.WorkflowQueryResponse:
     )
 
 
+class TestUserRegistryCredential(unittest.TestCase):
+    def test_valid_cred_preserves_path_scoped_registry(self):
+        credential = objects.UserRegistryCredential(
+            registry='nvcr.io/nvidia',
+            username='user',
+            auth='token',
+        )
+        workflow_config = mock.Mock()
+        workflow_config.credential_config.disable_registry_validation = []
+
+        with mock.patch(
+            'src.service.core.workflow.objects.common.registry_auth',
+            return_value=mock.Mock(status_code=200),
+        ) as registry_auth:
+            credential.valid_cred(workflow_config)
+
+        self.assertEqual(credential.registry, 'nvcr.io/nvidia')
+        registry_auth.assert_called_once_with('https://nvcr.io/v2/', 'user', 'token')
+
+
 class TestGetResourceNodeHash(unittest.TestCase):
 
     def test_get_resource_node_hash_empty_returns_hash_of_empty_string(self):

src/utils/connectors/postgres.py

@@ -1541,16 +1541,56 @@ def get_generic_cred(self, user: str, cred_name: str) -> Any:
 
     def get_registry_cred(self, user: str, registry: str) -> Any:
         """ Fetch docker credentials by registry name. """
+        registry = common.normalize_registry_scope(registry)
         select_data_cmd = PostgresSelectCommand(
             table='credential',
-            conditions=['user_name = %s', 'profile = %s'],
-            condition_args=[user, registry])
+            conditions=['user_name = %s', 'cred_type = %s', 'profile = %s'],
+            condition_args=[user, CredentialType.REGISTRY.value, registry])
         row = self.execute_fetch_command(*select_data_cmd.get_args())
         if row:
             return self.decrypt_credential(row[0])
         else:
             return None
 
+    def get_all_registry_creds(self, user: str) -> Dict[str, Dict[str, str]]:
+        """ Fetch all Docker registry credentials for a user by scope. """
+        select_data_cmd = PostgresSelectCommand(
+            table='credential',
+            conditions=['user_name = %s', 'cred_type = %s'],
+            condition_args=[user, CredentialType.REGISTRY.value])
+        rows = self.execute_fetch_command(*select_data_cmd.get_args())
+        registry_creds: Dict[str, Dict[str, str]] = {}
+        for row in rows:
+            if row.profile:
+                registry_scope = common.normalize_registry_scope(row.profile)
+                registry_creds[registry_scope] = self.decrypt_credential(row)
+        return registry_creds
+
+    def get_matching_registry_creds(
+        self,
+        user: str,
+        image_info: common.DockerImageInfo,
+    ) -> List[Tuple[str, Dict[str, str]]]:
+        """ Fetch all Docker registry credentials matching an image. """
+        select_data_cmd = PostgresSelectCommand(
+            table='credential',
+            conditions=['user_name = %s', 'cred_type = %s'],
+            condition_args=[user, CredentialType.REGISTRY.value])
+        rows = self.execute_fetch_command(*select_data_cmd.get_args())
+
+        registry_rows: Dict[str, List[Any]] = {}
+        for row in rows:
+            if row.profile:
+                registry_scope = common.normalize_registry_scope(row.profile)
+                registry_rows.setdefault(registry_scope, []).append(row)
+
+        return [
+            (registry_scope, self.decrypt_credential(row))
+            for registry_scope in common.matching_registry_scopes(
+                image_info, registry_rows.keys())
+            for row in registry_rows[registry_scope]
+        ]
+
     def get_workflow_service_url(self) -> str:
         """ Get the workflow service url. """
         service_config = self.get_service_configs()

src/utils/job/task.py

@@ -98,6 +98,9 @@ def create_login_dict(user: str,
     }
 
 
+def docker_auth(username: str, password: str) -> str:
+    return base64.b64encode(f'{username}:{password}'.encode('utf-8')).decode('utf-8')
+
 
 def create_config_dict(
     data_info: Mapping[str, credentials.StaticDataCredential | credentials.DefaultDataCredential],
@@ -2494,17 +2497,16 @@ def _get_image_secret_name(self, group_uid: str, name: str):
     def _get_registry_creds(self, user: str, workflow_config: connectors.WorkflowConfig):
         """ Got registry credentials for both user and osmo. """
         registry_creds_user = {}
-        registry_cred_cache: Dict[str, Any] = {}
+        registry_cred_map = self.database.get_all_registry_creds(user)
         for task in self.spec.tasks:
             image_info = common.docker_parse(task.image)
-            if image_info.host not in registry_cred_cache:
-                registry_cred_cache[image_info.host] = self.database.get_registry_cred(
-                    user, image_info.host)
-            payload = registry_cred_cache[image_info.host]
-            if payload:
-                auth_string = f'''{payload['username']}:{payload['auth']}'''
-                registry_creds_user[image_info.host] = \
-                    {'auth': base64.b64encode(auth_string.encode('utf-8')).decode('utf-8')}
+            for registry_scope in common.matching_registry_scopes(
+                image_info, registry_cred_map.keys()
+            ):
+                payload = registry_cred_map[registry_scope]
+                normalized_scope = common.normalize_registry_scope(registry_scope)
+                registry_creds_user[normalized_scope] = \
+                    {'auth': docker_auth(payload['username'], payload['auth'])}
 
         registry_cred_osmo = None
         osmo_cred = workflow_config.backend_images.credential
@@ -2514,11 +2516,10 @@ def _get_registry_creds(self, user: str, workflow_config: connectors.WorkflowCon
             and osmo_cred.username
             and osmo_cred.auth.get_secret_value()
         ):
-            auth_string = (
-                f'{osmo_cred.username}:{osmo_cred.auth.get_secret_value()}')
+            registry_scope = common.normalize_registry_scope(osmo_cred.registry)
             registry_cred_osmo = {
-                osmo_cred.registry: {
-                    'auth': base64.b64encode(auth_string.encode('utf-8')).decode('utf-8')
+                registry_scope: {
+                    'auth': docker_auth(osmo_cred.username, osmo_cred.auth.get_secret_value())
                 }
             }
         return registry_creds_user, registry_cred_osmo