import json from contextlib import contextmanager from types import SimpleNamespace from unittest.mock import MagicMock, Mock import pytest from flask import Response, request from mlflow.environment_variables import MLFLOW_ENABLE_WORKSPACES from mlflow.exceptions import MlflowException from mlflow.prompt.constants import IS_PROMPT_TAG_KEY from mlflow.protos.databricks_pb2 import RESOURCE_DOES_NOT_EXIST from mlflow.server import auth as auth_module from mlflow.server.auth.permissions import EDIT, MANAGE, NO_PERMISSIONS, READ, USE from mlflow.server.auth.routes import ( CREATE_PROMPTLAB_RUN, GET_ARTIFACT, GET_METRIC_HISTORY_BULK, GET_METRIC_HISTORY_BULK_INTERVAL, GET_MODEL_VERSION_ARTIFACT, GET_TRACE_ARTIFACT, GET_TRACE_ARTIFACT_V3, SEARCH_DATASETS, UPLOAD_ARTIFACT, ) from mlflow.server.auth.sqlalchemy_store import SqlAlchemyStore from mlflow.utils import workspace_context from tests.helper_functions import random_str def test_cleanup_workspace_permissions_handler(monkeypatch): mock_delete_workspace_perms = Mock() mock_delete_roles = Mock() monkeypatch.setattr( auth_module.store, "delete_workspace_permissions_for_workspace", mock_delete_workspace_perms, raising=True, ) monkeypatch.setattr( auth_module.store, "delete_roles_for_workspace", mock_delete_roles, raising=True, ) workspace_name = f"team-{random_str(10)}" with auth_module.app.test_request_context( f"/api/3.0/mlflow/workspaces/{workspace_name}", method="DELETE" ): request.view_args = {"workspace_name": workspace_name} response = Response(status=204) auth_module._after_request(response) mock_delete_workspace_perms.assert_called_once_with(workspace_name) mock_delete_roles.assert_called_once_with(workspace_name) def _create_workspace_response(workspace_name: str) -> Response: payload = {"workspace": {"name": workspace_name}} return Response(json.dumps(payload), status=201, content_type="application/json") def test_seed_default_workspace_roles_happy_path(monkeypatch): monkeypatch.setenv("MLFLOW_RBAC_SEED_DEFAULT_ROLES", "true") workspace_name = f"team-{random_str(10)}" created_roles: list[dict[str, object]] = [] added_perms: list[dict[str, object]] = [] def fake_create_role(name, workspace, description=None): role_id = len(created_roles) + 1 created_roles.append({ "id": role_id, "name": name, "workspace": workspace, "description": description, }) return SimpleNamespace(id=role_id, name=name, workspace=workspace) def fake_add_role_permission(role_id, resource_type, resource_pattern, permission): added_perms.append({ "role_id": role_id, "resource_type": resource_type, "resource_pattern": resource_pattern, "permission": permission, }) return SimpleNamespace(id=role_id) monkeypatch.setattr(auth_module.store, "create_role", fake_create_role, raising=True) monkeypatch.setattr( auth_module.store, "add_role_permission", fake_add_role_permission, raising=True ) with auth_module.app.test_request_context("/api/3.0/mlflow/workspaces", method="POST"): auth_module._seed_default_workspace_roles(_create_workspace_response(workspace_name)) names = [r["name"] for r in created_roles] assert names == ["admin", "user"] assert all(r["workspace"] == workspace_name for r in created_roles) # The simplified two-tier model lives in a single ``resource_type='workspace'`` # slot: ``admin`` carries MANAGE (admin grant), ``user`` carries USE (regular # member). The permission tier distinguishes the two without needing a separate # ``resource_type`` discriminant. assert [(p["resource_type"], p["permission"]) for p in added_perms] == [ ("workspace", MANAGE.name), ("workspace", USE.name), ] assert all(p["resource_pattern"] == "*" for p in added_perms) def test_seed_default_workspace_roles_disabled_skips_seeding(monkeypatch): # With seeding off, no roles are created. ``CreateWorkspace`` is gated to # super-admins so the creator already bypasses RBAC — there is nothing to # fall back to. monkeypatch.setenv("MLFLOW_RBAC_SEED_DEFAULT_ROLES", "false") workspace_name = f"team-{random_str(10)}" mock_create_role = Mock() mock_add_role_permission = Mock() mock_assign_role_to_user = Mock() mock_set_workspace_permission = Mock() monkeypatch.setattr(auth_module.store, "create_role", mock_create_role, raising=True) monkeypatch.setattr( auth_module.store, "add_role_permission", mock_add_role_permission, raising=True ) monkeypatch.setattr( auth_module.store, "assign_role_to_user", mock_assign_role_to_user, raising=True ) monkeypatch.setattr( auth_module.store, "set_workspace_permission", mock_set_workspace_permission, raising=True, ) with auth_module.app.test_request_context("/api/3.0/mlflow/workspaces", method="POST"): auth_module._seed_default_workspace_roles(_create_workspace_response(workspace_name)) mock_create_role.assert_not_called() mock_add_role_permission.assert_not_called() mock_assign_role_to_user.assert_not_called() mock_set_workspace_permission.assert_not_called() def test_seed_default_workspace_roles_admin_creation_fails_still_seeds_others(monkeypatch): # Best-effort seeding: a failure on one role doesn't block the rest. monkeypatch.setenv("MLFLOW_RBAC_SEED_DEFAULT_ROLES", "true") workspace_name = f"team-{random_str(10)}" def fake_create_role(name, workspace, description=None): if name == "admin": raise MlflowException("simulated admin role failure") return SimpleNamespace(id=10, name=name, workspace=workspace) mock_add_role_permission = Mock() monkeypatch.setattr(auth_module.store, "create_role", fake_create_role, raising=True) monkeypatch.setattr( auth_module.store, "add_role_permission", mock_add_role_permission, raising=True ) with auth_module.app.test_request_context("/api/3.0/mlflow/workspaces", method="POST"): auth_module._seed_default_workspace_roles(_create_workspace_response(workspace_name)) # ``user`` still got created (best-effort seeding). assert mock_add_role_permission.call_count == 1 def test_seed_default_workspace_roles_permission_add_fails_rolls_back_role(monkeypatch): # create_role succeeds but add_role_permission raises — the orphan role must be # deleted so the workspace doesn't end up with a named role that grants nothing. monkeypatch.setenv("MLFLOW_RBAC_SEED_DEFAULT_ROLES", "true") workspace_name = f"team-{random_str(10)}" def fake_create_role(name, workspace, description=None): return SimpleNamespace( id={"admin": 1, "user": 2}[name], name=name, workspace=workspace, ) def fake_add_role_permission(role_id, resource_type, resource_pattern, permission): if role_id == 1: # admin raise MlflowException("simulated add_role_permission failure") return SimpleNamespace(id=role_id) mock_delete_role = Mock() monkeypatch.setattr(auth_module.store, "create_role", fake_create_role, raising=True) monkeypatch.setattr( auth_module.store, "add_role_permission", fake_add_role_permission, raising=True ) monkeypatch.setattr(auth_module.store, "delete_role", mock_delete_role, raising=True) with auth_module.app.test_request_context("/api/3.0/mlflow/workspaces", method="POST"): auth_module._seed_default_workspace_roles(_create_workspace_response(workspace_name)) # Orphan admin role (id=1) was rolled back. mock_delete_role.assert_called_once_with(1) class _TrackingStore: def __init__( self, experiment_workspaces: dict[str, str], run_experiments: dict[str, str], trace_experiments: dict[str, str], experiment_names: dict[str, str] | None = None, logged_model_experiments: dict[str, str] | None = None, gateway_secret_workspaces: dict[str, str] | None = None, gateway_endpoint_workspaces: dict[str, str] | None = None, gateway_model_def_workspaces: dict[str, str] | None = None, engine=None, ManagedSessionMaker=None, ): self._experiment_workspaces = experiment_workspaces self._run_experiments = run_experiments self._trace_experiments = trace_experiments self._experiment_names = experiment_names or {} self._logged_model_experiments = logged_model_experiments or {} self._gateway_secret_workspaces = gateway_secret_workspaces or {} self._gateway_endpoint_workspaces = gateway_endpoint_workspaces or {} self._gateway_model_def_workspaces = gateway_model_def_workspaces or {} self.engine = engine self.ManagedSessionMaker = ManagedSessionMaker def get_experiment(self, experiment_id: str): if experiment_id not in self._experiment_workspaces: raise MlflowException( f"Experiment {experiment_id!r} not found", RESOURCE_DOES_NOT_EXIST ) return SimpleNamespace(workspace=self._experiment_workspaces[experiment_id]) def get_experiment_by_name(self, experiment_name: str): experiment_id = self._experiment_names.get(experiment_name) if experiment_id is None: return None return SimpleNamespace( experiment_id=experiment_id, workspace=self._experiment_workspaces[experiment_id], ) def get_run(self, run_id: str): return SimpleNamespace(info=SimpleNamespace(experiment_id=self._run_experiments[run_id])) def get_trace_info(self, request_id: str): return SimpleNamespace(experiment_id=self._trace_experiments[request_id]) def get_logged_model(self, model_id: str): experiment_id = self._logged_model_experiments[model_id] return SimpleNamespace(experiment_id=experiment_id) def get_secret_info(self, secret_id: str | None = None, secret_name: str | None = None): if secret_id: if secret_id not in self._gateway_secret_workspaces: raise MlflowException( f"GatewaySecret not found ({secret_id})", error_code=RESOURCE_DOES_NOT_EXIST, ) # Add workspace attribute so _get_resource_workspace can extract it return SimpleNamespace( secret_id=secret_id, workspace=self._gateway_secret_workspaces[secret_id] ) raise ValueError("Must provide secret_id or secret_name") def get_gateway_endpoint(self, endpoint_id: str | None = None, name: str | None = None): # For test simplicity we treat ``name`` as a synonym for ``endpoint_id`` # (our fixture data uses the same string for both). This mirrors how the # real store resolves a name → id lookup before returning the endpoint. if lookup_id := (endpoint_id or name): if lookup_id not in self._gateway_endpoint_workspaces: raise MlflowException( f"GatewayEndpoint not found ({lookup_id})", error_code=RESOURCE_DOES_NOT_EXIST, ) # Add workspace attribute so _get_resource_workspace can extract it return SimpleNamespace( endpoint_id=lookup_id, workspace=self._gateway_endpoint_workspaces[lookup_id] ) raise ValueError("Must provide endpoint_id or name") def get_gateway_model_definition( self, model_definition_id: str | None = None, name: str | None = None ): if model_definition_id: if model_definition_id not in self._gateway_model_def_workspaces: raise MlflowException( f"GatewayModelDefinition not found ({model_definition_id})", error_code=RESOURCE_DOES_NOT_EXIST, ) # Add workspace attribute so _get_resource_workspace can extract it return SimpleNamespace( model_definition_id=model_definition_id, workspace=self._gateway_model_def_workspaces[model_definition_id], ) raise ValueError("Must provide model_definition_id or name") def _create_mock_session(self): """Create a mock session that can query gateway SQL models.""" mock_session = MagicMock() def _filter_by_secret_id(secret_id): if secret_id in self._gateway_secret_workspaces: mock_result = MagicMock() mock_result.first.return_value = SimpleNamespace( workspace=self._gateway_secret_workspaces[secret_id] ) return mock_result mock_result = MagicMock() mock_result.first.return_value = None return mock_result def _filter_by_endpoint_id(endpoint_id): if endpoint_id in self._gateway_endpoint_workspaces: mock_result = MagicMock() mock_result.first.return_value = SimpleNamespace( workspace=self._gateway_endpoint_workspaces[endpoint_id] ) return mock_result mock_result = MagicMock() mock_result.first.return_value = None return mock_result def _filter_by_model_def_id(model_definition_id): if model_definition_id in self._gateway_model_def_workspaces: mock_result = MagicMock() mock_result.first.return_value = SimpleNamespace( workspace=self._gateway_model_def_workspaces[model_definition_id] ) return mock_result mock_result = MagicMock() mock_result.first.return_value = None return mock_result def _query(model_class): mock_query_result = MagicMock() # Mock the filter method to return different results based on the filter def _mock_filter(*args, **kwargs): if "secret_id" in kwargs: return _filter_by_secret_id(kwargs["secret_id"]) elif "endpoint_id" in kwargs: return _filter_by_endpoint_id(kwargs["endpoint_id"]) elif "model_definition_id" in kwargs: return _filter_by_model_def_id(kwargs["model_definition_id"]) return mock_query_result mock_query_result.filter = _mock_filter return mock_query_result mock_session.query = _query return mock_session def _create_mock_session_maker(self): """Create a mock ManagedSessionMaker context manager.""" @contextmanager def _mock_session_maker(): yield self._create_mock_session() return _mock_session_maker class _RegistryStore: def __init__(self, model_workspaces: dict[str, str], prompts: set[str] | None = None): self._model_workspaces = model_workspaces self._prompts = prompts or set() def get_registered_model(self, name: str): if name not in self._model_workspaces: raise MlflowException( f"Registered Model with name={name!r} not found", error_code=RESOURCE_DOES_NOT_EXIST, ) is_prompt = name in self._prompts return SimpleNamespace( workspace=self._model_workspaces[name], _is_prompt=lambda: is_prompt, ) @pytest.fixture def workspace_permission_setup(tmp_path, monkeypatch): monkeypatch.setenv(MLFLOW_ENABLE_WORKSPACES.name, "true") monkeypatch.setattr( auth_module, "auth_config", auth_module.auth_config._replace(default_permission=NO_PERMISSIONS.name), ) db_uri = f"sqlite:///{tmp_path / 'auth-store.db'}" auth_store = SqlAlchemyStore() auth_store.init_db(db_uri) monkeypatch.setattr(auth_module, "store", auth_store, raising=False) username = "alice" auth_store.create_user(username, "supersecurepassword", is_admin=False) tracking_store = _TrackingStore( experiment_workspaces={"exp-1": "team-a", "exp-2": "team-a", "1": "team-a"}, run_experiments={"run-1": "exp-1", "run-2": "exp-2"}, trace_experiments={"trace-1": "exp-1"}, experiment_names={"Primary Experiment": "exp-1"}, logged_model_experiments={"model-1": "exp-1"}, gateway_secret_workspaces={"secret-1": "team-a", "secret-2": "team-a"}, gateway_endpoint_workspaces={"endpoint-1": "team-a", "endpoint-2": "team-a"}, gateway_model_def_workspaces={"model-def-1": "team-a", "model-def-2": "team-a"}, engine=MagicMock(), # Mock engine for SQL model queries ) # Set ManagedSessionMaker after creating the store tracking_store.ManagedSessionMaker = tracking_store._create_mock_session_maker() monkeypatch.setattr(auth_module, "_get_tracking_store", lambda: tracking_store) registry_store = _RegistryStore({"model-xyz": "team-a"}) monkeypatch.setattr(auth_module, "_get_model_registry_store", lambda: registry_store) monkeypatch.setattr( auth_module, "authenticate_request", lambda: SimpleNamespace(username=username), ) auth_store.set_workspace_permission("team-a", username, MANAGE.name) with workspace_context.WorkspaceContext("team-a"): yield {"store": auth_store, "username": username} auth_store.engine.dispose() def _set_workspace_permission(store: SqlAlchemyStore, username: str, permission: str): """Replace the user's workspace grant on ``team-a`` with ``permission``. The ``workspace_permission_setup`` fixture pre-grants MANAGE so each test starts from a known authority; this helper rewrites that grant. ``permission == NO_PERMISSIONS`` is treated as "clear the grant" since the simplified model rejects NO_PERMISSIONS as a workspace-grant value — absence of a grant combined with ``default_permission=NO_PERMISSIONS`` produces the same deny semantics the explicit row used to provide. """ if permission == NO_PERMISSIONS.name: try: store.delete_workspace_permission("team-a", username) except MlflowException: pass return store.set_workspace_permission("team-a", username, permission) def test_filter_list_workspaces_includes_default_when_autogrant(monkeypatch): monkeypatch.setattr(auth_module, "sender_is_admin", lambda: False) auth = SimpleNamespace(username="alice") monkeypatch.setattr(auth_module, "authenticate_request", lambda: auth) monkeypatch.setattr( auth_module, "auth_config", auth_module.auth_config._replace( grant_default_workspace_access=True, default_permission=READ.name, ), raising=False, ) default_workspace = "team-default" monkeypatch.setattr(auth_module, "_get_workspace_store", lambda: None, raising=False) monkeypatch.setattr( auth_module, "get_default_workspace_optional", lambda *args, **kwargs: (SimpleNamespace(name=default_workspace), True), raising=False, ) class DummyStore: def list_accessible_workspace_names(self, username): return [] monkeypatch.setattr(auth_module, "store", DummyStore(), raising=False) response = Response( json.dumps({ "workspaces": [ {"name": default_workspace}, {"name": "other-workspace"}, ] }), mimetype="application/json", ) auth_module.filter_list_workspaces(response) payload = json.loads(response.get_data(as_text=True)) assert payload["workspaces"] == [{"name": default_workspace}] def test_filter_list_workspaces_filters_to_allowed(monkeypatch): monkeypatch.setattr(auth_module, "sender_is_admin", lambda: False) auth = SimpleNamespace(username="alice") monkeypatch.setattr(auth_module, "authenticate_request", lambda: auth) monkeypatch.setattr( auth_module, "auth_config", auth_module.auth_config._replace( grant_default_workspace_access=False, ), raising=False, ) class DummyStore: def list_accessible_workspace_names(self, username): return ["team-a"] monkeypatch.setattr(auth_module, "store", DummyStore(), raising=False) response = Response( json.dumps({"workspaces": [{"name": "team-a"}, {"name": "team-b"}]}), mimetype="application/json", ) auth_module.filter_list_workspaces(response) payload = json.loads(response.get_data(as_text=True)) assert [ws["name"] for ws in payload["workspaces"]] == ["team-a"] def test_list_workspaces_hides_workspace_with_only_synthetic_resource_grant(tmp_path, monkeypatch): # Pins the docs scenario: a user with only a per-resource grant (e.g. # ``(experiment, exp-456, EDIT)``) inside team-a does NOT see team-a in # ``mlflow.list_workspaces()`` — workspace visibility requires a # workspace-scoped role assignment, not a synthetic per-resource grant. # # The grant lives on Carol's synthetic ``__user___`` role with a # single ``(experiment, exp-456, EDIT)`` row — no ``(workspace, *)`` # grant. ``list_accessible_workspace_names`` should treat this as "no # workspace-level access" and filter team-a out. monkeypatch.setenv(MLFLOW_ENABLE_WORKSPACES.name, "true") monkeypatch.setattr(auth_module, "sender_is_admin", lambda: False) monkeypatch.setattr( auth_module, "auth_config", auth_module.auth_config._replace(grant_default_workspace_access=False), raising=False, ) db_uri = f"sqlite:///{tmp_path / 'auth-store.db'}" auth_store = SqlAlchemyStore() auth_store.init_db(db_uri) monkeypatch.setattr(auth_module, "store", auth_store, raising=False) auth_store.create_user("carol", "supersecurepassword", is_admin=False) # Direct per-resource grant via ``grant_user_permission`` — writes to # Carol's synthetic ``__user___`` role in team-a. This is exactly the # surface the admin UI's "Direct permissions" section sits on top of. with workspace_context.WorkspaceContext("team-a"): auth_store.grant_user_permission("carol", "experiment", "exp-456", "EDIT") monkeypatch.setattr( auth_module, "authenticate_request", lambda: SimpleNamespace(username="carol") ) response = Response( json.dumps({"workspaces": [{"name": "team-a"}, {"name": "team-b"}]}), mimetype="application/json", ) auth_module.filter_list_workspaces(response) payload = json.loads(response.get_data(as_text=True)) # protobuf-to-JSON omits empty repeated fields, so an empty list of # workspaces serializes as ``{}``. Both forms mean "Carol sees no # workspaces", which is what the docs claim. assert [ws["name"] for ws in payload.get("workspaces", [])] == [], ( "A direct-permission-only user must not see team-a in list_workspaces; " "only workspace-scoped role assignments confer visibility." ) auth_store.engine.dispose() def test_list_workspaces_filters_to_role_assigned_workspaces(tmp_path, monkeypatch): # End-to-end guard for the list_accessible_workspace_names fix: alice has NO # legacy workspace_permissions rows — her only workspace membership is via a # role assignment in ws-alpha. The ListWorkspaces filter must treat that role # assignment as workspace visibility and surface ws-alpha but not ws-beta. # Before the fix, the legacy-only query returned an empty set and alice saw # no workspaces in the UI. monkeypatch.setenv(MLFLOW_ENABLE_WORKSPACES.name, "true") monkeypatch.setattr(auth_module, "sender_is_admin", lambda: False) monkeypatch.setattr( auth_module, "auth_config", auth_module.auth_config._replace(grant_default_workspace_access=False), raising=False, ) db_uri = f"sqlite:///{tmp_path / 'auth-store.db'}" auth_store = SqlAlchemyStore() auth_store.init_db(db_uri) monkeypatch.setattr(auth_module, "store", auth_store, raising=False) alice = auth_store.create_user("alice", "supersecurepassword", is_admin=False) role = auth_store.create_role(name="viewer", workspace="ws-alpha") auth_store.add_role_permission(role.id, "experiment", "*", READ.name) auth_store.assign_role_to_user(alice.id, role.id) monkeypatch.setattr( auth_module, "authenticate_request", lambda: SimpleNamespace(username="alice") ) response = Response( json.dumps({"workspaces": [{"name": "ws-alpha"}, {"name": "ws-beta"}]}), mimetype="application/json", ) auth_module.filter_list_workspaces(response) payload = json.loads(response.get_data(as_text=True)) assert [ws["name"] for ws in payload["workspaces"]] == ["ws-alpha"] auth_store.engine.dispose() def test_validate_can_view_workspace_allows_default_autogrant(monkeypatch): monkeypatch.setenv(MLFLOW_ENABLE_WORKSPACES.name, "true") monkeypatch.setattr(auth_module, "sender_is_admin", lambda: False) auth = SimpleNamespace(username="alice") monkeypatch.setattr(auth_module, "authenticate_request", lambda: auth) monkeypatch.setattr( auth_module, "auth_config", auth_module.auth_config._replace( grant_default_workspace_access=True, default_permission=READ.name, ), raising=False, ) default_workspace = "team-default" monkeypatch.setattr(auth_module, "_get_workspace_store", lambda: None, raising=False) monkeypatch.setattr( auth_module, "get_default_workspace_optional", lambda *args, **kwargs: (SimpleNamespace(name=default_workspace), True), raising=False, ) class DummyStore: def list_accessible_workspace_names(self, username): return [] monkeypatch.setattr(auth_module, "store", DummyStore(), raising=False) with auth_module.app.test_request_context( f"/api/3.0/mlflow/workspaces/{default_workspace}", method="GET" ): request.view_args = {"workspace_name": default_workspace} assert auth_module.validate_can_view_workspace() with auth_module.app.test_request_context( "/api/3.0/mlflow/workspaces/other-team", method="GET" ): request.view_args = {"workspace_name": "other-team"} assert not auth_module.validate_can_view_workspace() def test_experiment_validators_allow_manage_permission(workspace_permission_setup): store = workspace_permission_setup["store"] username = workspace_permission_setup["username"] _set_workspace_permission(store, username, MANAGE.name) with auth_module.app.test_request_context( "/api/2.0/mlflow/experiments/get", method="GET", query_string={"experiment_id": "exp-1"} ): assert auth_module.validate_can_read_experiment() assert auth_module.validate_can_update_experiment() assert auth_module.validate_can_delete_experiment() assert auth_module.validate_can_manage_experiment() with auth_module.app.test_request_context( "/api/2.0/mlflow/experiments/get-by-name", method="GET", query_string={"experiment_name": "Primary Experiment"}, ): assert auth_module.validate_can_read_experiment_by_name() with workspace_context.WorkspaceContext("team-a"): assert auth_module.validate_can_create_experiment() def test_experiment_validators_allow_role_based_workspace_manage(workspace_permission_setup): # Grant MANAGE on the workspace via a role (with no legacy # ``workspace_permissions`` row). Pre-fix this returned 403 — the # workspace-level permission check only consulted the legacy table. # ``_workspace_permission`` now max-merges role-based grants. store = workspace_permission_setup["store"] username = workspace_permission_setup["username"] # Drop the legacy grant the fixture installs so we exercise the role path # in isolation. _set_workspace_permission(store, username, NO_PERMISSIONS.name) role = store.create_role(name="ws-admin", workspace="team-a") store.add_role_permission(role.id, "workspace", "*", MANAGE.name) user = store.get_user(username) store.assign_role_to_user(user.id, role.id) with auth_module.app.test_request_context( "/api/2.0/mlflow/experiments/get", method="GET", query_string={"experiment_id": "exp-1"} ): assert auth_module.validate_can_read_experiment() assert auth_module.validate_can_update_experiment() assert auth_module.validate_can_delete_experiment() assert auth_module.validate_can_manage_experiment() with workspace_context.WorkspaceContext("team-a"): assert auth_module.validate_can_create_experiment() def test_experiment_validators_workspace_use_allows_create_but_blocks_reads_on_others( workspace_permission_setup, ): # Workspace USE confers create + workspace access, not read on others' # resources — that needs an explicit per-resource grant. store = workspace_permission_setup["store"] username = workspace_permission_setup["username"] _set_workspace_permission(store, username, USE.name) with auth_module.app.test_request_context( "/api/2.0/mlflow/experiments/get", method="GET", query_string={"experiment_id": "exp-1"} ): assert not auth_module.validate_can_read_experiment() assert not auth_module.validate_can_update_experiment() assert not auth_module.validate_can_delete_experiment() assert not auth_module.validate_can_manage_experiment() with workspace_context.WorkspaceContext("team-a"): assert auth_module.validate_can_create_experiment() def test_workspace_permission_max_merges_legacy_and_role(workspace_permission_setup): # Operators mid-migration may have BOTH a legacy grant and a role grant. # The effective permission must be the higher of the two — neither side # should silently downgrade the other. store = workspace_permission_setup["store"] username = workspace_permission_setup["username"] # Legacy USE + role MANAGE → effective MANAGE. _set_workspace_permission(store, username, USE.name) role = store.create_role(name="ws-admin", workspace="team-a") store.add_role_permission(role.id, "workspace", "*", MANAGE.name) user = store.get_user(username) store.assign_role_to_user(user.id, role.id) with workspace_context.WorkspaceContext("team-a"): assert auth_module.validate_can_create_experiment() def test_use_workspace_permission_allows_create_but_blocks_reads_and_writes_on_others( workspace_permission_setup, ): store = workspace_permission_setup["store"] username = workspace_permission_setup["username"] _set_workspace_permission(store, username, USE.name) with workspace_context.WorkspaceContext("team-a"): assert auth_module.validate_can_create_experiment() assert auth_module.validate_can_create_registered_model() with auth_module.app.test_request_context( "/api/2.0/mlflow/experiments/get", method="GET", query_string={"experiment_id": "exp-1"} ): assert not auth_module.validate_can_read_experiment() assert not auth_module.validate_can_update_experiment() assert not auth_module.validate_can_delete_experiment() assert not auth_module.validate_can_manage_experiment() with auth_module.app.test_request_context( "/api/2.0/mlflow/experiments/get-by-name", method="GET", query_string={"experiment_name": "Primary Experiment"}, ): assert not auth_module.validate_can_read_experiment_by_name() with ( workspace_context.WorkspaceContext("team-a"), auth_module.app.test_request_context( "/api/2.0/mlflow/registered-models/get", method="GET", query_string={"name": "model-xyz"}, ), ): assert not auth_module.validate_can_read_registered_model() assert not auth_module.validate_can_update_registered_model() assert not auth_module.validate_can_delete_registered_model() assert not auth_module.validate_can_manage_registered_model() def test_no_permissions_blocks_create(workspace_permission_setup): # Without any access to the workspace, create is denied. store = workspace_permission_setup["store"] username = workspace_permission_setup["username"] _set_workspace_permission(store, username, NO_PERMISSIONS.name) with workspace_context.WorkspaceContext("team-a"): assert not auth_module.validate_can_create_experiment() assert not auth_module.validate_can_create_registered_model() def test_role_grant_workspace_use_allows_create(workspace_permission_setup, monkeypatch): # Workspace-wide USE grant ('workspace', '*', USE) confers create rights. store = workspace_permission_setup["store"] username = workspace_permission_setup["username"] user_id = store.get_user(username).id monkeypatch.setattr(auth_module, "sender_is_admin", lambda: False) # Drop the legacy grant the fixture installs so we exercise the role-only # path in isolation. _set_workspace_permission(store, username, NO_PERMISSIONS.name) role = store.create_role(name="ws-contributor", workspace="team-a") store.add_role_permission(role.id, "workspace", "*", USE.name) store.assign_role_to_user(user_id, role.id) with workspace_context.WorkspaceContext("team-a"): assert auth_module.validate_can_create_experiment() assert auth_module.validate_can_create_registered_model() def test_role_grant_resource_type_use_does_not_allow_create( workspace_permission_setup, monkeypatch ): # Resource-specific USE on ``experiment`` doesn't confer workspace-wide # create rights — only workspace-wide grants do. store = workspace_permission_setup["store"] username = workspace_permission_setup["username"] user_id = store.get_user(username).id monkeypatch.setattr(auth_module, "sender_is_admin", lambda: False) _set_workspace_permission(store, username, NO_PERMISSIONS.name) role = store.create_role(name="exp-user", workspace="team-a") store.add_role_permission(role.id, "experiment", "*", USE.name) store.assign_role_to_user(user_id, role.id) with workspace_context.WorkspaceContext("team-a"): assert not auth_module.validate_can_create_experiment() assert not auth_module.validate_can_create_registered_model() def test_experiment_artifact_proxy_validators_respect_permissions(workspace_permission_setup): store = workspace_permission_setup["store"] username = workspace_permission_setup["username"] _set_workspace_permission(store, username, MANAGE.name) with auth_module.app.test_request_context( "/ajax-api/2.0/mlflow-artifacts/artifacts/1/path", method="GET", ): request.view_args = {"artifact_path": "1/path"} assert auth_module.validate_can_read_experiment_artifact_proxy() assert auth_module.validate_can_update_experiment_artifact_proxy() assert auth_module.validate_can_delete_experiment_artifact_proxy() _set_workspace_permission(store, username, USE.name) with auth_module.app.test_request_context( "/ajax-api/2.0/mlflow-artifacts/artifacts/1/path", method="GET", ): request.view_args = {"artifact_path": "1/path"} assert not auth_module.validate_can_read_experiment_artifact_proxy() assert not auth_module.validate_can_update_experiment_artifact_proxy() assert not auth_module.validate_can_delete_experiment_artifact_proxy() def test_experiment_artifact_proxy_without_experiment_id_uses_workspace_permissions( workspace_permission_setup, ): store = workspace_permission_setup["store"] username = workspace_permission_setup["username"] _set_workspace_permission(store, username, USE.name) with auth_module.app.test_request_context( "/ajax-api/2.0/mlflow-artifacts/artifacts/uploads/path", method="GET", ): request.view_args = {"artifact_path": "uploads/path"} assert auth_module.validate_can_read_experiment_artifact_proxy() assert not auth_module.validate_can_update_experiment_artifact_proxy() def test_experiment_artifact_proxy_without_experiment_id_denied_without_workspace_permission( workspace_permission_setup, ): store = workspace_permission_setup["store"] username = workspace_permission_setup["username"] _set_workspace_permission(store, username, NO_PERMISSIONS.name) with auth_module.app.test_request_context( "/ajax-api/2.0/mlflow-artifacts/artifacts/uploads/path", method="GET", ): request.view_args = {"artifact_path": "uploads/path"} assert not auth_module.validate_can_read_experiment_artifact_proxy() def test_experiment_artifact_proxy_resolves_experiment_id_under_workspace_prefix( workspace_permission_setup, ): # In a non-default workspace the proxied artifact path is prefixed with # ``workspaces//``. The experiment id must still be resolved from the path so # per-experiment grants apply, instead of falling back to the workspace-tier grant # (which, for a USE member, has no can_update and would wrongly reject writes). store = workspace_permission_setup["store"] username = workspace_permission_setup["username"] # "DataScientist" shape: workspace USE (member, no can_update at the workspace tier) # plus an explicit experiment-level EDIT grant. _set_workspace_permission(store, username, USE.name) store.create_experiment_permission("1", username, EDIT.name) prefixed_path = "workspaces/team-a/1/run-1/artifacts/plots/x.png" with auth_module.app.test_request_context( f"/ajax-api/2.0/mlflow-artifacts/artifacts/{prefixed_path}", method="GET", ): request.view_args = {"artifact_path": prefixed_path} # EDIT on the experiment resolves through the workspace prefix -> reads and # writes are allowed even though the workspace-tier grant is only USE. assert auth_module.validate_can_read_experiment_artifact_proxy() assert auth_module.validate_can_update_experiment_artifact_proxy() # EDIT does not confer delete. assert not auth_module.validate_can_delete_experiment_artifact_proxy() def test_filter_experiment_ids_respects_workspace_permissions( workspace_permission_setup, monkeypatch ): store = workspace_permission_setup["store"] username = workspace_permission_setup["username"] monkeypatch.setattr(auth_module, "sender_is_admin", lambda: False) experiment_ids = ["exp-1", "exp-2"] assert auth_module.filter_experiment_ids(experiment_ids) == experiment_ids _set_workspace_permission(store, username, NO_PERMISSIONS.name) assert auth_module.filter_experiment_ids(experiment_ids) == [] def test_filter_experiment_ids_role_wildcard_grant(workspace_permission_setup, monkeypatch): # Role granting experiment(*) in the active workspace should include all experiments. store = workspace_permission_setup["store"] username = workspace_permission_setup["username"] user_id = store.get_user(username).id monkeypatch.setattr(auth_module, "sender_is_admin", lambda: False) # Start from NO_PERMISSIONS: workspace fallback would exclude everything. _set_workspace_permission(store, username, NO_PERMISSIONS.name) role = store.create_role(name="exp-reader", workspace="team-a") store.add_role_permission(role.id, "experiment", "*", "READ") store.assign_role_to_user(user_id, role.id) token = workspace_context.set_server_request_workspace("team-a") try: assert auth_module.filter_experiment_ids(["exp-1", "exp-2"]) == ["exp-1", "exp-2"] finally: workspace_context._WORKSPACE.reset(token) def test_filter_experiment_ids_role_specific_grant(workspace_permission_setup, monkeypatch): # Role granting a specific experiment id should include that id only (plus direct grants). store = workspace_permission_setup["store"] username = workspace_permission_setup["username"] user_id = store.get_user(username).id monkeypatch.setattr(auth_module, "sender_is_admin", lambda: False) _set_workspace_permission(store, username, NO_PERMISSIONS.name) role = store.create_role(name="exp-1-reader", workspace="team-a") store.add_role_permission(role.id, "experiment", "exp-1", "READ") store.assign_role_to_user(user_id, role.id) token = workspace_context.set_server_request_workspace("team-a") try: # Only exp-1 (via role); exp-2 is filtered out. assert auth_module.filter_experiment_ids(["exp-1", "exp-2"]) == ["exp-1"] finally: workspace_context._WORKSPACE.reset(token) def test_filter_experiment_ids_workspace_scope_role(workspace_permission_setup, monkeypatch): # Role with ('workspace', '*', USE) should grant read access to all experiments. store = workspace_permission_setup["store"] username = workspace_permission_setup["username"] user_id = store.get_user(username).id monkeypatch.setattr(auth_module, "sender_is_admin", lambda: False) _set_workspace_permission(store, username, NO_PERMISSIONS.name) role = store.create_role(name="ws-user", workspace="team-a") store.add_role_permission(role.id, "workspace", "*", "USE") store.assign_role_to_user(user_id, role.id) token = workspace_context.set_server_request_workspace("team-a") try: assert auth_module.filter_experiment_ids(["exp-1", "exp-2"]) == ["exp-1", "exp-2"] finally: workspace_context._WORKSPACE.reset(token) def test_run_validators_allow_manage_permission(workspace_permission_setup): store = workspace_permission_setup["store"] username = workspace_permission_setup["username"] _set_workspace_permission(store, username, MANAGE.name) with auth_module.app.test_request_context( "/api/2.0/mlflow/runs/get", method="GET", query_string={"run_id": "run-1"} ): assert auth_module.validate_can_read_run() assert auth_module.validate_can_update_run() assert auth_module.validate_can_delete_run() assert auth_module.validate_can_manage_run() def test_run_validators_workspace_use_blocks_reads_and_writes(workspace_permission_setup): store = workspace_permission_setup["store"] username = workspace_permission_setup["username"] _set_workspace_permission(store, username, USE.name) with auth_module.app.test_request_context( "/api/2.0/mlflow/runs/get", method="GET", query_string={"run_id": "run-1"} ): assert not auth_module.validate_can_read_run() assert not auth_module.validate_can_update_run() assert not auth_module.validate_can_delete_run() assert not auth_module.validate_can_manage_run() def test_create_model_version_source_read_blocks_cross_workspace( workspace_permission_setup, monkeypatch ): # Target model lives in team-a (user has MANAGE); source run/model live in team-b # (user has no access). The source-read check must block anchoring the model version # at a run/model the caller cannot read, even though they can update the target model. tracking_store = _TrackingStore( experiment_workspaces={"exp-a": "team-a", "exp-b": "team-b"}, run_experiments={"run-a": "exp-a", "run-b": "exp-b"}, trace_experiments={}, logged_model_experiments={"model-a": "exp-a", "model-b": "exp-b"}, ) monkeypatch.setattr(auth_module, "_get_tracking_store", lambda: tracking_store) with auth_module.app.test_request_context( "/api/2.0/mlflow/model-versions/create", method="POST", json={"name": "model-xyz", "source": "s3://bucket/x", "run_id": "run-b"}, ): assert not auth_module.validate_can_create_model_version() with auth_module.app.test_request_context( "/api/2.0/mlflow/model-versions/create", method="POST", json={"name": "model-xyz", "source": "s3://bucket/x", "model_id": "model-b"}, ): assert not auth_module.validate_can_create_model_version() # Same-workspace source (team-a) is allowed. with auth_module.app.test_request_context( "/api/2.0/mlflow/model-versions/create", method="POST", json={"name": "model-xyz", "source": "s3://bucket/x", "run_id": "run-a"}, ): assert auth_module.validate_can_create_model_version() # A truthy non-dict JSON body must be coerced to {} rather than raising # AttributeError on `.get(...)`. Bypass the target-model check (which reads # `name` from the body) to isolate the source-read coercion. monkeypatch.setattr( auth_module, "_validate_can_update_registered_model_or_prompt", lambda: True ) with auth_module.app.test_request_context( "/api/2.0/mlflow/model-versions/create", method="POST", json=[1], ): assert auth_module.validate_can_create_model_version() def test_logged_model_validators_respect_permissions(workspace_permission_setup): store = workspace_permission_setup["store"] username = workspace_permission_setup["username"] _set_workspace_permission(store, username, MANAGE.name) with auth_module.app.test_request_context( "/api/2.0/mlflow/logged-models/get", method="GET", query_string={"model_id": "model-1"}, ): assert auth_module.validate_can_read_logged_model() assert auth_module.validate_can_update_logged_model() assert auth_module.validate_can_delete_logged_model() assert auth_module.validate_can_manage_logged_model() _set_workspace_permission(store, username, USE.name) with auth_module.app.test_request_context( "/api/2.0/mlflow/logged-models/get", method="GET", query_string={"model_id": "model-1"}, ): assert not auth_module.validate_can_read_logged_model() assert not auth_module.validate_can_update_logged_model() assert not auth_module.validate_can_delete_logged_model() assert not auth_module.validate_can_manage_logged_model() def test_scorer_validators_use_workspace_permissions(workspace_permission_setup): store = workspace_permission_setup["store"] username = workspace_permission_setup["username"] _set_workspace_permission(store, username, MANAGE.name) with auth_module.app.test_request_context( "/api/3.0/mlflow/scorers/get", method="GET", query_string={"experiment_id": "exp-1", "name": "score-1"}, ): assert auth_module.validate_can_read_scorer() assert auth_module.validate_can_update_scorer() assert auth_module.validate_can_delete_scorer() assert auth_module.validate_can_manage_scorer() with auth_module.app.test_request_context( "/api/3.0/mlflow/scorers/permissions/create", method="POST", json={ "experiment_id": "exp-1", "scorer_name": "score-1", "username": "bob", "permission": "READ", }, ): assert auth_module.validate_can_manage_scorer_permission() def test_scorer_validators_workspace_use_blocks_reads_and_writes(workspace_permission_setup): store = workspace_permission_setup["store"] username = workspace_permission_setup["username"] _set_workspace_permission(store, username, USE.name) with auth_module.app.test_request_context( "/api/3.0/mlflow/scorers/get", method="GET", query_string={"experiment_id": "exp-1", "name": "score-1"}, ): assert not auth_module.validate_can_read_scorer() assert not auth_module.validate_can_update_scorer() assert not auth_module.validate_can_delete_scorer() assert not auth_module.validate_can_manage_scorer() with auth_module.app.test_request_context( "/api/3.0/mlflow/scorers/permissions/create", method="POST", json={ "experiment_id": "exp-1", "scorer_name": "score-1", "username": "bob", "permission": "READ", }, ): assert not auth_module.validate_can_manage_scorer_permission() def test_registered_model_validators_require_manage_for_writes(workspace_permission_setup): store = workspace_permission_setup["store"] username = workspace_permission_setup["username"] with workspace_context.WorkspaceContext("team-a"): _set_workspace_permission(store, username, MANAGE.name) with auth_module.app.test_request_context( "/api/2.0/mlflow/registered-models/get", method="GET", query_string={"name": "model-xyz"}, ): assert auth_module.validate_can_read_registered_model() assert auth_module.validate_can_update_registered_model() assert auth_module.validate_can_delete_registered_model() assert auth_module.validate_can_manage_registered_model() user = store.get_user(auth_module.authenticate_request().username) assert store.is_workspace_admin(user.id, "team-a") assert workspace_context.get_request_workspace() == "team-a" assert auth_module.validate_can_create_registered_model() _set_workspace_permission(store, username, USE.name) with auth_module.app.test_request_context( "/api/2.0/mlflow/registered-models/get", method="GET", query_string={"name": "model-xyz"}, ): assert not auth_module.validate_can_read_registered_model() assert not auth_module.validate_can_update_registered_model() assert not auth_module.validate_can_delete_registered_model() assert not auth_module.validate_can_manage_registered_model() # USE still confers create rights via creator-as-owner. assert auth_module.validate_can_create_registered_model() def test_prompt_validators_require_manage_for_writes(workspace_permission_setup, monkeypatch): store = workspace_permission_setup["store"] username = workspace_permission_setup["username"] registry_store = _RegistryStore({"prompt-xyz": "team-a"}, prompts={"prompt-xyz"}) monkeypatch.setattr(auth_module, "_get_model_registry_store", lambda: registry_store) with workspace_context.WorkspaceContext("team-a"): _set_workspace_permission(store, username, MANAGE.name) with auth_module.app.test_request_context( "/api/2.0/mlflow/registered-models/get", method="GET", query_string={"name": "prompt-xyz"}, ): assert auth_module.validate_can_read_prompt() assert auth_module.validate_can_update_prompt() assert auth_module.validate_can_delete_prompt() assert auth_module.validate_can_manage_prompt() _set_workspace_permission(store, username, USE.name) with auth_module.app.test_request_context( "/api/2.0/mlflow/registered-models/get", method="GET", query_string={"name": "prompt-xyz"}, ): assert not auth_module.validate_can_read_prompt() assert not auth_module.validate_can_update_prompt() assert not auth_module.validate_can_delete_prompt() assert not auth_module.validate_can_manage_prompt() def test_prompt_dispatch_routes_request_by_is_prompt_tag(workspace_permission_setup, monkeypatch): # Shared registered-model route resolves to the prompt resource_type only when # the entity is a prompt; the dispatching wrappers pick via # `_get_permission_from_registered_model_or_prompt_name()` (single fetch + `._is_prompt()`). store = workspace_permission_setup["store"] username = workspace_permission_setup["username"] registry_store = _RegistryStore( {"prompt-xyz": "team-a", "model-xyz": "team-a"}, prompts={"prompt-xyz"}, ) monkeypatch.setattr(auth_module, "_get_model_registry_store", lambda: registry_store) # Grant prompt READ only — NOT registered_model READ. role = store.create_role(name="prompt-reader", workspace="team-a") store.add_role_permission(role.id, "prompt", "prompt-xyz", "READ") user = store.get_user(username) store.assign_role_to_user(user.id, role.id) _set_workspace_permission(store, username, NO_PERMISSIONS.name) with workspace_context.WorkspaceContext("team-a"): # The prompt name maps to the prompt validator and succeeds. with auth_module.app.test_request_context( "/api/2.0/mlflow/registered-models/get", method="GET", query_string={"name": "prompt-xyz"}, ): assert auth_module._validate_can_read_registered_model_or_prompt() # A non-prompt name maps to the registered_model validator and FAILS: # pins cross-resource isolation (prompt grant ≠ registered_model grant). with auth_module.app.test_request_context( "/api/2.0/mlflow/registered-models/get", method="GET", query_string={"name": "model-xyz"}, ): assert not auth_module._validate_can_read_registered_model_or_prompt() def test_registered_model_grant_does_not_satisfy_prompt_request( workspace_permission_setup, monkeypatch ): """The inverse of the dispatch test: a `(registered_model, foo, READ)` grant must NOT satisfy a request for prompt `foo` after the resource_type split. """ store = workspace_permission_setup["store"] username = workspace_permission_setup["username"] registry_store = _RegistryStore({"foo": "team-a"}, prompts={"foo"}) monkeypatch.setattr(auth_module, "_get_model_registry_store", lambda: registry_store) role = store.create_role(name="rm-reader", workspace="team-a") store.add_role_permission(role.id, "registered_model", "foo", "READ") user = store.get_user(username) store.assign_role_to_user(user.id, role.id) _set_workspace_permission(store, username, NO_PERMISSIONS.name) with workspace_context.WorkspaceContext("team-a"): with auth_module.app.test_request_context( "/api/2.0/mlflow/registered-models/get", method="GET", query_string={"name": "foo"}, ): # registered_model grant must not leak into the prompt namespace. assert not auth_module.validate_can_read_prompt() def test_request_targets_prompt_is_registry_driven_not_body_driven( workspace_permission_setup, monkeypatch ): # Spoofing regression: classification reads the persisted tag, not the body. # Otherwise `(prompt, foo, MANAGE)` could flip the namespace on a non-CREATE # registered-model route via a spoofed body tag. # ``foo`` exists as a regular registered model, not a prompt. registry_store = _RegistryStore({"foo": "team-a"}, prompts=set()) monkeypatch.setattr(auth_module, "_get_model_registry_store", lambda: registry_store) with workspace_context.WorkspaceContext("team-a"): # Spoofed body tag on a non-CREATE route must NOT classify ``foo`` as a prompt. with auth_module.app.test_request_context( "/api/2.0/mlflow/registered-models/delete", method="DELETE", json={ "name": "foo", "tags": [{"key": "mlflow.prompt.is_prompt", "value": "true"}], }, ): assert not auth_module._request_targets_prompt() # Without the spoofed tag, the persisted-state lookup still says # registered_model — body tags have no impact on classification. with auth_module.app.test_request_context( "/api/2.0/mlflow/registered-models/delete", method="DELETE", json={"name": "foo"}, ): assert not auth_module._request_targets_prompt() def test_request_targets_prompt_persisted_prompt_classifies_true( workspace_permission_setup, monkeypatch ): # Mirror of the spoofing regression: persisted `_is_prompt() == True` routes # to the prompt validator regardless of the body. registry_store = _RegistryStore({"foo": "team-a"}, prompts={"foo"}) monkeypatch.setattr(auth_module, "_get_model_registry_store", lambda: registry_store) with workspace_context.WorkspaceContext("team-a"): with auth_module.app.test_request_context( "/api/2.0/mlflow/registered-models/get", method="GET", query_string={"name": "foo"}, ): assert auth_module._request_targets_prompt() def test_request_targets_prompt_unknown_entity_falls_back_to_registered_model( workspace_permission_setup, monkeypatch ): # Non-existent entity → False (registered-model path surfaces the 404). # Body-tag spoof can't force the prompt namespace because the body is ignored. registry_store = _RegistryStore({}) # empty — every lookup raises 404 monkeypatch.setattr(auth_module, "_get_model_registry_store", lambda: registry_store) with workspace_context.WorkspaceContext("team-a"): with auth_module.app.test_request_context( "/api/2.0/mlflow/registered-models/create", method="POST", json={ "name": "new-prompt", "tags": [{"key": "mlflow.prompt.is_prompt", "value": "true"}], }, ): assert not auth_module._request_targets_prompt() def test_request_targets_prompt_propagates_unexpected_errors( workspace_permission_setup, monkeypatch ): # Non-`RESOURCE_DOES_NOT_EXIST` errors must propagate; silencing them would # quietly route every request down the registered-model path. class _BrokenRegistryStore: def get_registered_model(self, name): raise RuntimeError("registry store backend is down") monkeypatch.setattr(auth_module, "_get_model_registry_store", lambda: _BrokenRegistryStore()) with workspace_context.WorkspaceContext("team-a"): with auth_module.app.test_request_context( "/api/2.0/mlflow/registered-models/get", method="GET", query_string={"name": "foo"}, ): with pytest.raises(RuntimeError, match="registry store backend is down"): auth_module._request_targets_prompt() def test_filter_search_registered_models_uses_prompt_grant_for_prompt_rows( workspace_permission_setup, monkeypatch ): # A user holding only ``(prompt, foo, READ)`` previously had prompt ``foo`` # silently filtered out of ``SearchRegisteredModels`` results because the # filter checked the ``registered_model`` namespace exclusively. With the # per-row classify, the prompt grant satisfies the prompt row. monkeypatch.setattr(auth_module, "sender_is_admin", lambda: False) store = workspace_permission_setup["store"] username = workspace_permission_setup["username"] _set_workspace_permission(store, username, NO_PERMISSIONS.name) role = store.create_role(name="prompt-reader", workspace="team-a") store.add_role_permission(role.id, "prompt", "foo", READ.name) store.assign_role_to_user(store.get_user(username).id, role.id) payload = json.dumps({ "registered_models": [ {"name": "foo", "tags": [{"key": IS_PROMPT_TAG_KEY, "value": "true"}]}, {"name": "bar", "tags": []}, ], "next_page_token": "", }) flask_resp = Response(payload, mimetype="application/json") with auth_module.app.test_request_context( "/api/2.0/mlflow/registered-models/search", method="GET", query_string={"max_results": "100"}, ): auth_module.filter_search_registered_models(flask_resp) out = json.loads(flask_resp.get_data(as_text=True)) names = [rm["name"] for rm in out.get("registered_models", [])] # Prompt ``foo`` is kept (grant satisfies prompt namespace); ``bar`` is filtered out. assert names == ["foo"] def test_filter_search_registered_models_does_not_satisfy_prompt_with_rm_grant( workspace_permission_setup, monkeypatch ): # Inverse direction: a ``(registered_model, foo, READ)`` grant must NOT # leak through and make a prompt row readable. Pins cross-namespace # isolation on the response-filtering path. monkeypatch.setattr(auth_module, "sender_is_admin", lambda: False) store = workspace_permission_setup["store"] username = workspace_permission_setup["username"] _set_workspace_permission(store, username, NO_PERMISSIONS.name) role = store.create_role(name="rm-reader", workspace="team-a") store.add_role_permission(role.id, "registered_model", "foo", READ.name) store.assign_role_to_user(store.get_user(username).id, role.id) payload = json.dumps({ "registered_models": [ {"name": "foo", "tags": [{"key": IS_PROMPT_TAG_KEY, "value": "true"}]}, ], "next_page_token": "", }) flask_resp = Response(payload, mimetype="application/json") with auth_module.app.test_request_context( "/api/2.0/mlflow/registered-models/search", method="GET", query_string={"max_results": "100"}, ): auth_module.filter_search_registered_models(flask_resp) out = json.loads(flask_resp.get_data(as_text=True)) assert out.get("registered_models", []) == [] def test_filter_search_model_versions_uses_prompt_grant_for_prompt_versions( workspace_permission_setup, monkeypatch ): # Same gap on ``SearchModelVersions``: prompt versions carry the # ``mlflow.prompt.is_prompt`` tag and must be checked against ``prompt`` # grants, not ``registered_model`` grants. monkeypatch.setattr(auth_module, "sender_is_admin", lambda: False) store = workspace_permission_setup["store"] username = workspace_permission_setup["username"] _set_workspace_permission(store, username, NO_PERMISSIONS.name) role = store.create_role(name="prompt-reader", workspace="team-a") store.add_role_permission(role.id, "prompt", "foo", READ.name) store.assign_role_to_user(store.get_user(username).id, role.id) payload = json.dumps({ "model_versions": [ {"name": "foo", "tags": [{"key": IS_PROMPT_TAG_KEY, "value": "true"}]}, {"name": "bar", "tags": []}, ], }) flask_resp = Response(payload, mimetype="application/json") with auth_module.app.test_request_context( "/api/2.0/mlflow/model-versions/search", method="GET" ): auth_module.filter_search_model_versions(flask_resp) out = json.loads(flask_resp.get_data(as_text=True)) names = [mv["name"] for mv in out.get("model_versions", [])] assert names == ["foo"] def test_rename_registered_model_permission_sweeps_prompt_namespace( workspace_permission_setup, ): # Renaming a prompt must propagate to ``(prompt, old_name, ...)`` grants. # Without sweeping both namespaces, the rename leaves those grants orphaned # under the old name and creates nothing under the new one. store = workspace_permission_setup["store"] username = workspace_permission_setup["username"] store.grant_user_permission(username, "prompt", "foo", READ.name) # Confirm the seed grant landed where we expect. user_id = store.get_user(username).id before = { (rp.resource_type, rp.resource_pattern) for role in store.list_user_roles(user_id) for rp in role.permissions } assert ("prompt", "foo") in before with auth_module.app.test_request_context( "/api/2.0/mlflow/registered-models/rename", method="POST", json={"name": "foo", "new_name": "bar"}, ): auth_module.rename_registered_model_permission(Response(status=200)) after = { (rp.resource_type, rp.resource_pattern) for role in store.list_user_roles(user_id) for rp in role.permissions } assert ("prompt", "foo") not in after assert ("prompt", "bar") in after def test_delete_can_manage_registered_model_permission_sweeps_prompt_namespace( workspace_permission_setup, ): # Deleting a prompt must sweep its ``(prompt, name, ...)`` grants. # Without the prompt-side delete, those rows would leak permanently. store = workspace_permission_setup["store"] username = workspace_permission_setup["username"] store.grant_user_permission(username, "prompt", "foo", READ.name) user_id = store.get_user(username).id before = { (rp.resource_type, rp.resource_pattern) for role in store.list_user_roles(user_id) for rp in role.permissions } assert ("prompt", "foo") in before with auth_module.app.test_request_context( "/api/2.0/mlflow/registered-models/delete", method="DELETE", json={"name": "foo"}, ): auth_module.delete_can_manage_registered_model_permission(Response(status=200)) after = { (rp.resource_type, rp.resource_pattern) for role in store.list_user_roles(user_id) for rp in role.permissions } assert ("prompt", "foo") not in after def test_set_can_manage_registered_model_permission_grants_prompt_for_prompt_entity( workspace_permission_setup, ): # ``CreateRegisteredModel`` is shared with prompt creation. When the # created entity is a prompt, the creator-default MANAGE grant must land # in the ``prompt`` namespace — otherwise the prompt-side validators # immediately lock the creator out of their own prompt. store = workspace_permission_setup["store"] username = workspace_permission_setup["username"] user_id = store.get_user(username).id flask_resp = Response( json.dumps({ "registered_model": { "name": "my-prompt", "tags": [{"key": IS_PROMPT_TAG_KEY, "value": "true"}], } }), mimetype="application/json", ) with auth_module.app.test_request_context( "/api/2.0/mlflow/registered-models/create", method="POST", json={"name": "my-prompt"}, ): auth_module.set_can_manage_registered_model_permission(flask_resp) grants = { (rp.resource_type, rp.resource_pattern, rp.permission) for role in store.list_user_roles(user_id) for rp in role.permissions } assert ("prompt", "my-prompt", MANAGE.name) in grants assert ("registered_model", "my-prompt", MANAGE.name) not in grants def test_set_can_manage_registered_model_permission_grants_registered_model_for_plain_entity( workspace_permission_setup, ): # Inverse: a non-prompt registered model still grants in the # ``registered_model`` namespace — pins that the new classification path # didn't accidentally flip the default for normal models. store = workspace_permission_setup["store"] username = workspace_permission_setup["username"] user_id = store.get_user(username).id flask_resp = Response( json.dumps({"registered_model": {"name": "my-model", "tags": []}}), mimetype="application/json", ) with auth_module.app.test_request_context( "/api/2.0/mlflow/registered-models/create", method="POST", json={"name": "my-model"}, ): auth_module.set_can_manage_registered_model_permission(flask_resp) grants = { (rp.resource_type, rp.resource_pattern, rp.permission) for role in store.list_user_roles(user_id) for rp in role.permissions } assert ("registered_model", "my-model", MANAGE.name) in grants assert ("prompt", "my-model", MANAGE.name) not in grants def test_filter_search_registered_models_classifies_refetched_rows( workspace_permission_setup, monkeypatch ): # The initial filter pass works on protos; if it doesn't fill # ``max_results``, the loop refetches more rows as ORM # ``RegisteredModel`` entities. Those ORM rows have ``.tags`` that hide # the ``mlflow.prompt.is_prompt`` key, so naive ``_proto_is_prompt`` on # them would misclassify every prompt as a registered_model. Pins that # ``_entity_is_prompt`` dispatches to ``_is_prompt()`` on the ORM side. from mlflow.entities.model_registry import RegisteredModel from mlflow.entities.model_registry.registered_model_tag import RegisteredModelTag from mlflow.store.entities import PagedList from mlflow.utils.search_utils import SearchUtils monkeypatch.setattr(auth_module, "sender_is_admin", lambda: False) store = workspace_permission_setup["store"] username = workspace_permission_setup["username"] _set_workspace_permission(store, username, NO_PERMISSIONS.name) role = store.create_role(name="prompt-reader", workspace="team-a") store.add_role_permission(role.id, "prompt", "refetched-prompt", READ.name) store.assign_role_to_user(store.get_user(username).id, role.id) # Initial response is empty + has a next_page_token so the refetch loop # runs. Then a fake registry returns one prompt + one registered_model. refetched_rows = [ RegisteredModel( name="refetched-prompt", tags=[RegisteredModelTag(key=IS_PROMPT_TAG_KEY, value="true")], ), RegisteredModel(name="refetched-model", tags=[]), ] # First refetch returns the seed rows; subsequent calls return an empty # page so the loop terminates instead of spinning on the same fake page. calls = {"count": 0} def fake_search(**_kwargs): calls["count"] += 1 return PagedList(refetched_rows if calls["count"] == 1 else [], token=None) fake_registry = SimpleNamespace(search_registered_models=fake_search) monkeypatch.setattr(auth_module, "_get_model_registry_store", lambda: fake_registry) # ``SearchUtils.parse_start_offset_from_page_token`` requires a real # base64-encoded JSON token; use the project's helper so the loop's # offset-bookkeeping doesn't reject our seed. seed_token = SearchUtils.create_page_token(1).decode("utf-8") flask_resp = Response( json.dumps({"registered_models": [], "next_page_token": seed_token}), mimetype="application/json", ) with auth_module.app.test_request_context( "/api/2.0/mlflow/registered-models/search", method="GET", query_string={"max_results": "10"}, ): auth_module.filter_search_registered_models(flask_resp) out = json.loads(flask_resp.get_data(as_text=True)) names = [rm["name"] for rm in out.get("registered_models", [])] # The refetched prompt row is kept (prompt grant satisfies it); the # plain registered_model row is filtered out. assert names == ["refetched-prompt"] def test_delete_can_manage_registered_model_permission_rejects_missing_name( workspace_permission_setup, ): # ``request.get_json(silent=True)`` returns ``None`` on missing / # unparsable bodies; the guard must surface a clean 400 instead of a # ``TypeError`` -> 500. with auth_module.app.test_request_context( "/api/2.0/mlflow/registered-models/delete", method="DELETE", data="", # empty body content_type="application/json", ): with pytest.raises(MlflowException, match="Missing value for required parameter 'name'"): auth_module.delete_can_manage_registered_model_permission(Response(status=200)) def test_rename_registered_model_permission_rejects_missing_fields( workspace_permission_setup, ): # Missing ``name`` / ``new_name`` must raise INVALID_PARAMETER_VALUE # rather than silently forwarding ``None`` to # ``rename_grants_for_resource`` where it would corrupt grants. with auth_module.app.test_request_context( "/api/2.0/mlflow/registered-models/rename", method="POST", json={"name": "foo"}, # no new_name ): with pytest.raises(MlflowException, match="Missing value for required parameter"): auth_module.rename_registered_model_permission(Response(status=200)) def test_validate_can_view_workspace_requires_access(workspace_permission_setup): store = workspace_permission_setup["store"] username = workspace_permission_setup["username"] with auth_module.app.test_request_context( "/api/3.0/mlflow/workspaces/team-a", method="GET", ): request.view_args = {"workspace_name": "team-a"} assert auth_module.validate_can_view_workspace() store.delete_workspace_permission("team-a", username) with auth_module.app.test_request_context( "/api/3.0/mlflow/workspaces/team-a", method="GET", ): request.view_args = {"workspace_name": "team-a"} assert not auth_module.validate_can_view_workspace() def test_run_artifact_validators_use_workspace_permissions(workspace_permission_setup): with auth_module.app.test_request_context( GET_ARTIFACT, method="GET", query_string={"run_id": "run-1"}, ): assert auth_module.validate_can_read_run_artifact() with auth_module.app.test_request_context( UPLOAD_ARTIFACT, method="POST", query_string={"run_id": "run-1"}, ): assert auth_module.validate_can_update_run_artifact() def test_model_version_artifact_validator_uses_workspace_permissions(workspace_permission_setup): with auth_module.app.test_request_context( GET_MODEL_VERSION_ARTIFACT, method="GET", query_string={"name": "model-xyz"}, ): assert auth_module.validate_can_read_model_version_artifact() def test_metric_history_bulk_validator_uses_workspace_permissions(workspace_permission_setup): with auth_module.app.test_request_context( GET_METRIC_HISTORY_BULK, method="GET", query_string=[("run_id", "run-1"), ("run_id", "run-2")], ): assert auth_module.validate_can_read_metric_history_bulk() def test_metric_history_bulk_interval_validator_uses_workspace_permissions( workspace_permission_setup, ): with auth_module.app.test_request_context( GET_METRIC_HISTORY_BULK_INTERVAL, method="GET", query_string=[ ("run_ids", "run-1"), ("run_ids", "run-2"), ("metric_key", "loss"), ], ): assert auth_module.validate_can_read_metric_history_bulk_interval() def test_search_datasets_validator_uses_workspace_permissions(workspace_permission_setup): with auth_module.app.test_request_context( SEARCH_DATASETS, method="POST", json={"experiment_ids": ["exp-1", "exp-2"]}, ): assert auth_module.validate_can_search_datasets() def test_create_promptlab_run_validator_uses_workspace_permissions(workspace_permission_setup): with auth_module.app.test_request_context( CREATE_PROMPTLAB_RUN, method="POST", json={"experiment_id": "exp-2"}, ): assert auth_module.validate_can_create_promptlab_run() @pytest.mark.parametrize("path", [GET_TRACE_ARTIFACT, GET_TRACE_ARTIFACT_V3]) def test_trace_artifact_validator_uses_workspace_permissions(workspace_permission_setup, path): with auth_module.app.test_request_context( path, method="GET", query_string={"request_id": "trace-1"}, ): assert auth_module.validate_can_read_trace_artifact() def test_experiment_artifact_proxy_without_workspaces_falls_back_to_default(monkeypatch): monkeypatch.setenv(MLFLOW_ENABLE_WORKSPACES.name, "false") monkeypatch.setattr( auth_module, "auth_config", auth_module.auth_config._replace(default_permission=READ.name), raising=False, ) monkeypatch.setattr( auth_module, "authenticate_request", lambda: SimpleNamespace(username="carol"), ) with auth_module.app.test_request_context( "/ajax-api/2.0/mlflow-artifacts/artifacts/uploads/path", method="GET", ): request.view_args = {"artifact_path": "uploads/path"} assert auth_module.validate_can_read_experiment_artifact_proxy() def test_run_artifact_validators_denied_without_workspace_permission(workspace_permission_setup): store = workspace_permission_setup["store"] username = workspace_permission_setup["username"] _set_workspace_permission(store, username, NO_PERMISSIONS.name) with auth_module.app.test_request_context( GET_ARTIFACT, method="GET", query_string={"run_id": "run-1"}, ): assert not auth_module.validate_can_read_run_artifact() with auth_module.app.test_request_context( UPLOAD_ARTIFACT, method="POST", query_string={"run_id": "run-1"}, ): assert not auth_module.validate_can_update_run_artifact() def test_model_version_artifact_validator_denied_without_workspace_permission( workspace_permission_setup, ): store = workspace_permission_setup["store"] username = workspace_permission_setup["username"] _set_workspace_permission(store, username, NO_PERMISSIONS.name) with auth_module.app.test_request_context( GET_MODEL_VERSION_ARTIFACT, method="GET", query_string={"name": "model-xyz"}, ): assert not auth_module.validate_can_read_model_version_artifact() def test_metric_history_bulk_validator_denied_without_workspace_permission( workspace_permission_setup, ): store = workspace_permission_setup["store"] username = workspace_permission_setup["username"] _set_workspace_permission(store, username, NO_PERMISSIONS.name) with auth_module.app.test_request_context( GET_METRIC_HISTORY_BULK, method="GET", query_string=[("run_id", "run-1"), ("run_id", "run-2")], ): assert not auth_module.validate_can_read_metric_history_bulk() def test_metric_history_bulk_interval_validator_denied_without_workspace_permission( workspace_permission_setup, ): store = workspace_permission_setup["store"] username = workspace_permission_setup["username"] _set_workspace_permission(store, username, NO_PERMISSIONS.name) with auth_module.app.test_request_context( GET_METRIC_HISTORY_BULK_INTERVAL, method="GET", query_string=[ ("run_ids", "run-1"), ("run_ids", "run-2"), ("metric_key", "loss"), ], ): assert not auth_module.validate_can_read_metric_history_bulk_interval() def test_search_datasets_validator_denied_without_workspace_permission( workspace_permission_setup, ): store = workspace_permission_setup["store"] username = workspace_permission_setup["username"] _set_workspace_permission(store, username, NO_PERMISSIONS.name) with auth_module.app.test_request_context( SEARCH_DATASETS, method="POST", json={"experiment_ids": ["exp-1", "exp-2"]}, ): assert not auth_module.validate_can_search_datasets() def test_create_promptlab_run_validator_denied_without_workspace_permission( workspace_permission_setup, ): store = workspace_permission_setup["store"] username = workspace_permission_setup["username"] _set_workspace_permission(store, username, NO_PERMISSIONS.name) with auth_module.app.test_request_context( CREATE_PROMPTLAB_RUN, method="POST", json={"experiment_id": "exp-2"}, ): assert not auth_module.validate_can_create_promptlab_run() @pytest.mark.parametrize("path", [GET_TRACE_ARTIFACT, GET_TRACE_ARTIFACT_V3]) def test_trace_artifact_validator_denied_without_workspace_permission( workspace_permission_setup, path ): store = workspace_permission_setup["store"] username = workspace_permission_setup["username"] _set_workspace_permission(store, username, NO_PERMISSIONS.name) with auth_module.app.test_request_context( path, method="GET", query_string={"request_id": "trace-1"}, ): assert not auth_module.validate_can_read_trace_artifact() def test_cross_workspace_access_denied(workspace_permission_setup, monkeypatch): tracking_store = _TrackingStore( experiment_workspaces={"exp-other-ws": "team-b"}, run_experiments={"run-other-ws": "exp-other-ws"}, trace_experiments={}, ) monkeypatch.setattr(auth_module, "_get_tracking_store", lambda: tracking_store) with auth_module.app.test_request_context( "/api/2.0/mlflow/experiments/get", method="GET", query_string={"experiment_id": "exp-other-ws"}, ): assert not auth_module.validate_can_read_experiment() assert not auth_module.validate_can_update_experiment() assert not auth_module.validate_can_delete_experiment() with auth_module.app.test_request_context( "/api/2.0/mlflow/runs/get", method="GET", query_string={"run_id": "run-other-ws"}, ): assert not auth_module.validate_can_read_run() assert not auth_module.validate_can_update_run() def test_cross_workspace_registered_model_access_denied(workspace_permission_setup, monkeypatch): registry_store = _RegistryStore({"model-other-ws": "team-b"}) monkeypatch.setattr(auth_module, "_get_model_registry_store", lambda: registry_store) with auth_module.app.test_request_context( "/api/2.0/mlflow/registered-models/get", method="GET", query_string={"name": "model-other-ws"}, ): assert not auth_module.validate_can_read_registered_model() assert not auth_module.validate_can_update_registered_model() assert not auth_module.validate_can_delete_registered_model() def test_explicit_experiment_permission_overrides_workspace( workspace_permission_setup, ): store = workspace_permission_setup["store"] username = workspace_permission_setup["username"] _set_workspace_permission(store, username, NO_PERMISSIONS.name) store.create_experiment_permission("exp-1", username, READ.name) with auth_module.app.test_request_context( "/api/2.0/mlflow/experiments/get", method="GET", query_string={"experiment_id": "exp-1"}, ): assert auth_module.validate_can_read_experiment() assert not auth_module.validate_can_update_experiment() with auth_module.app.test_request_context( "/api/2.0/mlflow/experiments/get", method="GET", query_string={"experiment_id": "exp-2"}, ): assert not auth_module.validate_can_read_experiment() def test_cross_workspace_gateway_secret_access_denied(workspace_permission_setup, monkeypatch): tracking_store = _TrackingStore( experiment_workspaces={"exp-1": "team-a"}, run_experiments={}, trace_experiments={}, gateway_secret_workspaces={"secret-other-ws": "team-b"}, engine=MagicMock(), ) tracking_store.ManagedSessionMaker = tracking_store._create_mock_session_maker() monkeypatch.setattr(auth_module, "_get_tracking_store", lambda: tracking_store) with auth_module.app.test_request_context( "/api/3.0/mlflow/gateway/secrets/get", method="GET", query_string={"secret_id": "secret-other-ws"}, ): assert not auth_module.validate_can_read_gateway_secret() assert not auth_module.validate_can_update_gateway_secret() assert not auth_module.validate_can_delete_gateway_secret() def test_cross_workspace_gateway_endpoint_access_denied(workspace_permission_setup, monkeypatch): tracking_store = _TrackingStore( experiment_workspaces={"exp-1": "team-a"}, run_experiments={}, trace_experiments={}, gateway_endpoint_workspaces={"endpoint-other-ws": "team-b"}, engine=MagicMock(), ) tracking_store.ManagedSessionMaker = tracking_store._create_mock_session_maker() monkeypatch.setattr(auth_module, "_get_tracking_store", lambda: tracking_store) with auth_module.app.test_request_context( "/api/3.0/mlflow/gateway/endpoints/get", method="GET", query_string={"endpoint_id": "endpoint-other-ws"}, ): assert not auth_module.validate_can_read_gateway_endpoint() assert not auth_module.validate_can_update_gateway_endpoint() assert not auth_module.validate_can_delete_gateway_endpoint() def test_cross_workspace_gateway_model_definition_access_denied( workspace_permission_setup, monkeypatch ): tracking_store = _TrackingStore( experiment_workspaces={"exp-1": "team-a"}, run_experiments={}, trace_experiments={}, gateway_model_def_workspaces={"model-def-other-ws": "team-b"}, engine=MagicMock(), ) tracking_store.ManagedSessionMaker = tracking_store._create_mock_session_maker() monkeypatch.setattr(auth_module, "_get_tracking_store", lambda: tracking_store) with auth_module.app.test_request_context( "/api/3.0/mlflow/gateway/model-definitions/get", method="GET", query_string={"model_definition_id": "model-def-other-ws"}, ): assert not auth_module.validate_can_read_gateway_model_definition() assert not auth_module.validate_can_update_gateway_model_definition() assert not auth_module.validate_can_delete_gateway_model_definition() def test_workspace_permission_required_for_gateway_creation(workspace_permission_setup): store = workspace_permission_setup["store"] username = workspace_permission_setup["username"] # Remove workspace permission _set_workspace_permission(store, username, NO_PERMISSIONS.name) with auth_module.app.test_request_context( "/api/3.0/mlflow/gateway/endpoints/create", method="POST", json={"name": "test-endpoint", "model_configs": []}, ): assert not auth_module.validate_can_create_gateway_endpoint() with auth_module.app.test_request_context( "/api/3.0/mlflow/gateway/model-definitions/create", method="POST", json={ "name": "test-model", "secret_id": "secret-1", "provider": "openai", "model_name": "gpt-4", }, ): assert not auth_module.validate_can_create_gateway_model_definition() # Restore workspace permission store.set_workspace_permission("team-a", username, MANAGE.name) with auth_module.app.test_request_context( "/api/3.0/mlflow/gateway/endpoints/create", method="POST", json={"name": "test-endpoint", "model_configs": []}, ): assert auth_module.validate_can_create_gateway_endpoint() def test_prompt_optimization_job_validators_use_workspace_permissions( workspace_permission_setup, monkeypatch ): store = workspace_permission_setup["store"] username = workspace_permission_setup["username"] # Mock get_job to return a job associated with exp-1 (in team-a) mock_job = SimpleNamespace(params='{"experiment_id": "exp-1"}') monkeypatch.setattr(auth_module, "get_job", lambda job_id: mock_job) _set_workspace_permission(store, username, MANAGE.name) with auth_module.app.test_request_context( "/api/3.0/mlflow/prompt-optimization/jobs/get", method="GET", query_string={"job_id": "job-1"}, ): assert auth_module.validate_can_read_prompt_optimization_job() assert auth_module.validate_can_update_prompt_optimization_job() assert auth_module.validate_can_delete_prompt_optimization_job() def test_prompt_optimization_job_validators_workspace_use_blocks_reads_and_writes( workspace_permission_setup, monkeypatch ): # Job auth gates on parent experiment permission; workspace USE no longer folds. store = workspace_permission_setup["store"] username = workspace_permission_setup["username"] mock_job = SimpleNamespace(params='{"experiment_id": "exp-1"}') monkeypatch.setattr(auth_module, "get_job", lambda job_id: mock_job) _set_workspace_permission(store, username, USE.name) with auth_module.app.test_request_context( "/api/3.0/mlflow/prompt-optimization/jobs/get", method="GET", query_string={"job_id": "job-1"}, ): assert not auth_module.validate_can_read_prompt_optimization_job() assert not auth_module.validate_can_update_prompt_optimization_job() assert not auth_module.validate_can_delete_prompt_optimization_job() def test_prompt_optimization_job_validators_denied_without_workspace_permission( workspace_permission_setup, monkeypatch ): store = workspace_permission_setup["store"] username = workspace_permission_setup["username"] # Mock get_job to return a job associated with exp-1 (in team-a) mock_job = SimpleNamespace(params='{"experiment_id": "exp-1"}') monkeypatch.setattr(auth_module, "get_job", lambda job_id: mock_job) _set_workspace_permission(store, username, NO_PERMISSIONS.name) with auth_module.app.test_request_context( "/api/3.0/mlflow/prompt-optimization/jobs/get", method="GET", query_string={"job_id": "job-1"}, ): assert not auth_module.validate_can_read_prompt_optimization_job() assert not auth_module.validate_can_update_prompt_optimization_job() assert not auth_module.validate_can_delete_prompt_optimization_job() def test_graphql_permission_functions_use_workspace_permissions(workspace_permission_setup): store = workspace_permission_setup["store"] username = workspace_permission_setup["username"] _set_workspace_permission(store, username, MANAGE.name) # Test experiment permission assert auth_module._graphql_can_read_experiment("exp-1", username) # Test run permission (inherits from experiment) assert auth_module._graphql_can_read_run("run-1", username) # Test registered model permission assert auth_module._graphql_can_read_model("model-xyz", username) def test_graphql_permission_functions_denied_without_workspace_permission( workspace_permission_setup, ): store = workspace_permission_setup["store"] username = workspace_permission_setup["username"] _set_workspace_permission(store, username, NO_PERMISSIONS.name) # Test experiment permission denied assert not auth_module._graphql_can_read_experiment("exp-1", username) # Test run permission denied (inherits from experiment) assert not auth_module._graphql_can_read_run("run-1", username) # Test registered model permission denied assert not auth_module._graphql_can_read_model("model-xyz", username) def test_cross_workspace_graphql_access_denied(workspace_permission_setup, monkeypatch): # User has MANAGE in team-a but tries to access resources in team-b tracking_store = _TrackingStore( experiment_workspaces={"exp-other-ws": "team-b"}, run_experiments={"run-other-ws": "exp-other-ws"}, trace_experiments={}, ) monkeypatch.setattr(auth_module, "_get_tracking_store", lambda: tracking_store) registry_store = _RegistryStore({"model-other-ws": "team-b"}) monkeypatch.setattr(auth_module, "_get_model_registry_store", lambda: registry_store) username = workspace_permission_setup["username"] # Should be denied access to resources in team-b assert not auth_module._graphql_can_read_experiment("exp-other-ws", username) assert not auth_module._graphql_can_read_run("run-other-ws", username) assert not auth_module._graphql_can_read_model("model-other-ws", username) # ============================================================================= # Role-based permission coverage for gateway resources # ============================================================================= # # The fixture grants workspace MANAGE by default. These tests first strip that # grant (set to NO_PERMISSIONS) so the only path to a positive permission is # the role assignment being exercised. That isolates the role-based resolver # from the legacy workspace_permissions fallback. def _assign_role_with_permission( store: SqlAlchemyStore, username: str, workspace: str, resource_type: str, permission: str ) -> None: """Create a role in ``workspace`` with a wildcard grant of ``permission`` on ``resource_type``, and assign ``username`` to it. Using ``random_str`` keeps the role names unique so multiple calls within a single test don't collide on the (workspace, name) unique constraint. """ role = store.create_role(name=random_str(), workspace=workspace) store.add_role_permission(role.id, resource_type, "*", permission) user = store.get_user(username) store.assign_role_to_user(user.id, role.id) # ---- Gateway endpoint: role-based permission levels ---- @pytest.mark.parametrize( ("granted", "expected_read", "expected_delete", "expected_manage"), [ ("READ", True, False, False), ("USE", True, False, False), ("EDIT", True, False, False), ("MANAGE", True, True, True), ], ) def test_role_grant_on_gateway_endpoint_gates_validator_capabilities( workspace_permission_setup, granted, expected_read, expected_delete, expected_manage ): """A role grant at permission level ``granted`` exposes exactly the capabilities that level implies on the endpoint validators — no more, no less. Catches regressions where a validator starts accepting a weaker permission than it should (or refuses a stronger one). """ store = workspace_permission_setup["store"] username = workspace_permission_setup["username"] # Strip the default workspace MANAGE so the only positive grant is the role. _set_workspace_permission(store, username, NO_PERMISSIONS.name) _assign_role_with_permission(store, username, "team-a", "gateway_endpoint", granted) with auth_module.app.test_request_context( "/api/3.0/mlflow/gateway/endpoints/get", method="GET", query_string={"endpoint_id": "endpoint-1"}, ): assert auth_module.validate_can_read_gateway_endpoint() is expected_read assert auth_module.validate_can_delete_gateway_endpoint() is expected_delete assert auth_module.validate_can_manage_gateway_endpoint() is expected_manage def test_role_grant_read_on_gateway_endpoint_does_not_permit_use( workspace_permission_setup, ): """Regression guard specific to the bug class the user called out: a user with only READ on a gateway endpoint should not be able to *invoke* it. USE is a stricter capability than READ and has its own validator. """ store = workspace_permission_setup["store"] username = workspace_permission_setup["username"] _set_workspace_permission(store, username, NO_PERMISSIONS.name) _assign_role_with_permission(store, username, "team-a", "gateway_endpoint", "READ") # _validate_gateway_use_permission looks up the endpoint by name, resolves # the endpoint id, then checks ``can_use`` via the permission resolver. with auth_module.app.test_request_context("/"): assert auth_module._validate_gateway_use_permission("endpoint-1", username) is False def test_role_grant_use_on_gateway_endpoint_permits_use(workspace_permission_setup): store = workspace_permission_setup["store"] username = workspace_permission_setup["username"] _set_workspace_permission(store, username, NO_PERMISSIONS.name) _assign_role_with_permission(store, username, "team-a", "gateway_endpoint", "USE") with auth_module.app.test_request_context("/"): assert auth_module._validate_gateway_use_permission("endpoint-1", username) is True @pytest.mark.parametrize( ("granted", "expected_can_use"), [ ("READ", False), # READ does not imply USE. ("USE", True), ("EDIT", True), # EDIT implies USE. ("MANAGE", True), # MANAGE implies USE. ], ) def test_role_grant_permission_level_determines_use_capability( workspace_permission_setup, granted, expected_can_use ): """Parametrized matrix for the USE capability specifically. READ should NOT let the user invoke; every stronger permission should. """ store = workspace_permission_setup["store"] username = workspace_permission_setup["username"] _set_workspace_permission(store, username, NO_PERMISSIONS.name) _assign_role_with_permission(store, username, "team-a", "gateway_endpoint", granted) with auth_module.app.test_request_context("/"): assert ( auth_module._validate_gateway_use_permission("endpoint-1", username) is expected_can_use ) # ---- Workspace-wide role grants on gateway resources ---- @pytest.mark.parametrize( ("granted", "expected_read", "expected_manage"), [ # USE doesn't fold into resource lookups; MANAGE does. ("USE", False, False), ("MANAGE", True, True), ], ) def test_role_workspace_wide_grant_folds_for_manage_only_on_gateway_endpoints( workspace_permission_setup, granted, expected_read, expected_manage ): store = workspace_permission_setup["store"] username = workspace_permission_setup["username"] _set_workspace_permission(store, username, NO_PERMISSIONS.name) _assign_role_with_permission(store, username, "team-a", "workspace", granted) with auth_module.app.test_request_context( "/api/3.0/mlflow/gateway/endpoints/get", method="GET", query_string={"endpoint_id": "endpoint-1"}, ): assert auth_module.validate_can_read_gateway_endpoint() is expected_read assert auth_module.validate_can_delete_gateway_endpoint() is expected_manage assert auth_module.validate_can_manage_gateway_endpoint() is expected_manage @pytest.mark.parametrize( ("granted", "expected_use"), [ # USE doesn't fold into resource lookups; MANAGE does. ("USE", False), ("MANAGE", True), ], ) def test_role_workspace_wide_grant_invocation_tier_dependent( workspace_permission_setup, granted, expected_use ): store = workspace_permission_setup["store"] username = workspace_permission_setup["username"] _set_workspace_permission(store, username, NO_PERMISSIONS.name) _assign_role_with_permission(store, username, "team-a", "workspace", granted) with auth_module.app.test_request_context("/"): assert auth_module._validate_gateway_use_permission("endpoint-1", username) is expected_use # ---- Gateway secret and model definition parity ---- @pytest.mark.parametrize( ("granted", "expected_read", "expected_delete"), [ ("READ", True, False), ("EDIT", True, False), ("MANAGE", True, True), ], ) def test_role_grant_on_gateway_secret_gates_validator( workspace_permission_setup, granted, expected_read, expected_delete ): store = workspace_permission_setup["store"] username = workspace_permission_setup["username"] _set_workspace_permission(store, username, NO_PERMISSIONS.name) _assign_role_with_permission(store, username, "team-a", "gateway_secret", granted) with auth_module.app.test_request_context( "/api/3.0/mlflow/gateway/secrets/get", method="GET", query_string={"secret_id": "secret-1"}, ): assert auth_module.validate_can_read_gateway_secret() is expected_read assert auth_module.validate_can_delete_gateway_secret() is expected_delete @pytest.mark.parametrize( ("granted", "expected_read", "expected_delete"), [ ("READ", True, False), ("EDIT", True, False), ("MANAGE", True, True), ], ) def test_role_grant_on_gateway_model_definition_gates_validator( workspace_permission_setup, granted, expected_read, expected_delete ): store = workspace_permission_setup["store"] username = workspace_permission_setup["username"] _set_workspace_permission(store, username, NO_PERMISSIONS.name) _assign_role_with_permission(store, username, "team-a", "gateway_model_definition", granted) with auth_module.app.test_request_context( "/api/3.0/mlflow/gateway/model-definitions/get", method="GET", query_string={"model_definition_id": "model-def-1"}, ): assert auth_module.validate_can_read_gateway_model_definition() is expected_read assert auth_module.validate_can_delete_gateway_model_definition() is expected_delete # ---- Cross-workspace isolation for role-based gateway grants ---- def test_role_in_other_workspace_does_not_grant_gateway_endpoint_access( workspace_permission_setup, ): """A role in team-b with MANAGE on gateway_endpoints must not grant any access when resolving an endpoint that belongs to team-a. The resolver scopes role permissions to the role's workspace. """ store = workspace_permission_setup["store"] username = workspace_permission_setup["username"] _set_workspace_permission(store, username, NO_PERMISSIONS.name) # Role with MANAGE in team-b — should NOT apply to team-a endpoints. _assign_role_with_permission(store, username, "team-b", "gateway_endpoint", "MANAGE") with auth_module.app.test_request_context( "/api/3.0/mlflow/gateway/endpoints/get", method="GET", query_string={"endpoint_id": "endpoint-1"}, # endpoint-1 is in team-a. ): assert auth_module.validate_can_read_gateway_endpoint() is False assert auth_module.validate_can_manage_gateway_endpoint() is False def test_role_in_other_workspace_does_not_grant_gateway_use(workspace_permission_setup): store = workspace_permission_setup["store"] username = workspace_permission_setup["username"] _set_workspace_permission(store, username, NO_PERMISSIONS.name) _assign_role_with_permission(store, username, "team-b", "gateway_endpoint", "USE") with auth_module.app.test_request_context("/"): # endpoint-1 is in team-a; role grant is in team-b. assert auth_module._validate_gateway_use_permission("endpoint-1", username) is False # ---- Multi-role union: best grant wins ---- def test_role_union_best_permission_wins_for_gateway_endpoint(workspace_permission_setup): """Two roles: one grants READ, the other grants MANAGE. Validator should reflect the max (MANAGE). """ store = workspace_permission_setup["store"] username = workspace_permission_setup["username"] _set_workspace_permission(store, username, NO_PERMISSIONS.name) _assign_role_with_permission(store, username, "team-a", "gateway_endpoint", "READ") _assign_role_with_permission(store, username, "team-a", "gateway_endpoint", "MANAGE") with auth_module.app.test_request_context( "/api/3.0/mlflow/gateway/endpoints/get", method="GET", query_string={"endpoint_id": "endpoint-1"}, ): assert auth_module.validate_can_manage_gateway_endpoint() is True # ============================================================================= # Authorization for role management endpoints (Batch 5) # ============================================================================= # # Four validators guard the role endpoints: # - validate_can_manage_roles: create/update/delete role, add/remove/update # role_permission, assign/unassign role. Super admin OR workspace admin # in the resolved workspace. # - validate_can_view_roles: get_role, list_role_permissions. Super admin # OR any role assignment in the resolved workspace. # - validate_can_list_roles: list_roles. Super admin unconditionally; for # non-admins the request must scope to a workspace where the caller holds # at least one role. # - validate_can_view_user_roles: list_user_roles. Super admin, the target # themselves, or a workspace admin over any workspace the target is in. # # _get_role_workspace_from_request resolves the workspace from role_id, # role_permission_id, or a literal ``workspace`` param. These tests exercise # all three shapes and every actor x endpoint combination. @pytest.fixture def role_auth_setup(tmp_path, monkeypatch): monkeypatch.setenv(MLFLOW_ENABLE_WORKSPACES.name, "true") monkeypatch.setattr( auth_module, "auth_config", auth_module.auth_config._replace(default_permission=NO_PERMISSIONS.name), ) db_uri = f"sqlite:///{tmp_path / 'auth-store.db'}" auth_store = SqlAlchemyStore() auth_store.init_db(db_uri) monkeypatch.setattr(auth_module, "store", auth_store, raising=False) auth_store.create_user("super_admin", "supersecurepassword", is_admin=True) for name in ("ws_admin_foo", "ws_admin_bar", "ws_member_foo", "outsider"): auth_store.create_user(name, "supersecurepassword", is_admin=False) admin_role_foo = auth_store.create_role(name="admin-foo", workspace="foo") auth_store.add_role_permission(admin_role_foo.id, "workspace", "*", MANAGE.name) auth_store.assign_role_to_user(auth_store.get_user("ws_admin_foo").id, admin_role_foo.id) admin_role_bar = auth_store.create_role(name="admin-bar", workspace="bar") auth_store.add_role_permission(admin_role_bar.id, "workspace", "*", MANAGE.name) auth_store.assign_role_to_user(auth_store.get_user("ws_admin_bar").id, admin_role_bar.id) member_role_foo = auth_store.create_role(name="member-foo", workspace="foo") auth_store.add_role_permission(member_role_foo.id, "experiment", "*", READ.name) auth_store.assign_role_to_user(auth_store.get_user("ws_member_foo").id, member_role_foo.id) role_foo = auth_store.create_role(name="target-foo", workspace="foo") role_bar = auth_store.create_role(name="target-bar", workspace="bar") rp_foo = auth_store.add_role_permission(role_foo.id, "experiment", "*", READ.name) rp_bar = auth_store.add_role_permission(role_bar.id, "experiment", "*", READ.name) def login_as(username: str) -> None: monkeypatch.setattr( auth_module, "authenticate_request", lambda: SimpleNamespace(username=username), ) yield { "store": auth_store, "login_as": login_as, "role_foo_id": role_foo.id, "role_bar_id": role_bar.id, "role_permission_foo_id": rp_foo.id, "role_permission_bar_id": rp_bar.id, } auth_store.engine.dispose() def _request_context_for_shape(shape, role_auth_setup, workspace): match shape: case "role_id": role_id = ( role_auth_setup["role_foo_id"] if workspace == "foo" else role_auth_setup["role_bar_id"] ) return auth_module.app.test_request_context( "/api/3.0/mlflow/roles/get", method="GET", query_string={"role_id": str(role_id)}, ) case "role_permission_id": rp_id = ( role_auth_setup["role_permission_foo_id"] if workspace == "foo" else role_auth_setup["role_permission_bar_id"] ) return auth_module.app.test_request_context( "/api/3.0/mlflow/roles/permissions/update", method="PATCH", json={"role_permission_id": rp_id, "permission": READ.name}, ) case "workspace": return auth_module.app.test_request_context( "/api/3.0/mlflow/roles/create", method="POST", json={"name": "new-role", "workspace": workspace}, ) case _: raise ValueError(f"Unknown shape: {shape}") # Authorization matrices are exercised with a single request shape (role_id); # shape-resolution itself is covered independently below so we don't multiply # every actor-case by three shape-cases. @pytest.mark.parametrize( ("actor", "workspace", "expected"), [ # Super admin short-circuits regardless of workspace — one case suffices. ("super_admin", "foo", True), # Outsider has no role anywhere — one case suffices. ("outsider", "foo", False), # Workspace admins manage only their own workspace. ("ws_admin_foo", "foo", True), ("ws_admin_foo", "bar", False), ("ws_admin_bar", "foo", False), ("ws_admin_bar", "bar", True), # Plain role membership is not enough to manage — needs workspace MANAGE. ("ws_member_foo", "foo", False), ("ws_member_foo", "bar", False), ], ) def test_validate_can_manage_roles_authorization(role_auth_setup, actor, workspace, expected): role_auth_setup["login_as"](actor) with _request_context_for_shape("role_id", role_auth_setup, workspace): assert auth_module.validate_can_manage_roles() is expected @pytest.mark.parametrize("workspace", ["foo", "bar"]) @pytest.mark.parametrize("shape", ["role_id", "role_permission_id", "workspace"]) def test_manage_roles_resolves_workspace_from_each_shape(role_auth_setup, shape, workspace): # Sanity check that _get_role_workspace_from_request correctly dispatches # on every request shape. Use ws_admin_foo — their answer differs by # workspace, so an incorrectly resolved (or swapped) workspace flips the # result and the test fails. role_auth_setup["login_as"]("ws_admin_foo") expected = workspace == "foo" with _request_context_for_shape(shape, role_auth_setup, workspace): assert auth_module.validate_can_manage_roles() is expected @pytest.mark.parametrize( ("actor", "workspace", "expected"), [ ("super_admin", "foo", True), ("outsider", "foo", False), ("ws_admin_foo", "foo", True), ("ws_admin_foo", "bar", False), ("ws_admin_bar", "foo", False), ("ws_admin_bar", "bar", True), # Unlike manage, a plain workspace member can view roles. ("ws_member_foo", "foo", True), ("ws_member_foo", "bar", False), ], ) def test_validate_can_view_roles_authorization(role_auth_setup, actor, workspace, expected): role_auth_setup["login_as"](actor) with _request_context_for_shape("role_id", role_auth_setup, workspace): assert auth_module.validate_can_view_roles() is expected @pytest.mark.parametrize( ("actor", "expected"), [ ("super_admin", True), # Any non-admin is denied regardless of their workspace memberships — # one representative non-admin is enough. ("ws_admin_foo", False), ], ) def test_validate_can_list_roles_unscoped_is_super_admin_only(role_auth_setup, actor, expected): # No workspace param: only super admins may list every role in the system. role_auth_setup["login_as"](actor) with auth_module.app.test_request_context("/api/3.0/mlflow/roles/list", method="GET"): assert auth_module.validate_can_list_roles() is expected @pytest.mark.parametrize( ("actor", "workspace", "expected"), [ ("super_admin", "foo", True), ("outsider", "foo", False), ("ws_admin_foo", "foo", True), ("ws_admin_foo", "bar", False), ("ws_admin_bar", "foo", False), ("ws_admin_bar", "bar", True), ("ws_member_foo", "foo", True), ("ws_member_foo", "bar", False), ], ) def test_validate_can_list_roles_workspace_scoped(role_auth_setup, actor, workspace, expected): role_auth_setup["login_as"](actor) with auth_module.app.test_request_context( "/api/3.0/mlflow/roles/list", method="GET", query_string={"workspace": workspace} ): assert auth_module.validate_can_list_roles() is expected def test_validate_can_list_roles_blank_workspace_denied_for_non_admin(role_auth_setup): # Blank workspace param hits a *different* branch from the missing-param # case: validate_can_list_roles checks ``workspace.strip()`` and denies # rather than raising, unlike _get_role_workspace_from_request which would # raise on blank workspace. Kept as a guard for that specific branch. role_auth_setup["login_as"]("ws_admin_foo") with auth_module.app.test_request_context( "/api/3.0/mlflow/roles/list", method="GET", query_string={"workspace": " "}, ): assert auth_module.validate_can_list_roles() is False @pytest.mark.parametrize( ("actor", "workspaces", "expected"), [ # Super admin lists across any combination unconditionally. ("super_admin", ["foo", "bar"], True), # Workspace admin must hold a role in *every* requested workspace. ("ws_admin_foo", ["foo"], True), ("ws_admin_foo", ["foo", "bar"], False), # not present in bar ("ws_admin_foo", ["foo", "foo"], True), # duplicate is fine # Member of foo can list foo, but not foo + bar. ("ws_member_foo", ["foo"], True), ("ws_member_foo", ["foo", "bar"], False), # Outsider can list nothing. ("outsider", ["foo"], False), ("outsider", ["foo", "bar"], False), ], ) def test_validate_can_list_roles_multi_workspace(role_auth_setup, actor, workspaces, expected): role_auth_setup["login_as"](actor) with auth_module.app.test_request_context( "/api/3.0/mlflow/roles/list", method="GET", query_string=[("workspace", w) for w in workspaces], ): assert auth_module.validate_can_list_roles() is expected # Listing users is scoped to workspace membership (the review-queue assignment UI # needs the roster; assigning still requires experiment MANAGE), so the roster # isn't leaked across workspaces. Super admin is omitted: ``_before_request`` # short-circuits via ``sender_is_admin`` before the validator is reached. @pytest.mark.parametrize( ("actor", "workspace", "expected"), [ # Workspace-wide grant carrying can_use (USE/MANAGE) → may list the roster. ("ws_admin_foo", "foo", True), # Isolation: an admin of another workspace can't list users in this one. ("ws_admin_foo", "bar", False), ("ws_admin_bar", "foo", False), # A plain experiment-level grant is not a workspace-wide grant → denied. ("ws_member_foo", "foo", False), # No grant anywhere. ("outsider", "foo", False), ], ) def test_validate_can_list_users_workspace_scoped(role_auth_setup, actor, workspace, expected): role_auth_setup["login_as"](actor) token = workspace_context.set_server_request_workspace(workspace) try: with auth_module.app.test_request_context("/api/2.0/mlflow/users/list", method="GET"): assert auth_module.validate_can_list_users() is expected finally: workspace_context._WORKSPACE.reset(token) def test_validate_can_list_users_allows_any_user_without_workspaces(role_auth_setup, monkeypatch): # With workspaces disabled there is no isolation boundary, so any authenticated # user may list the roster (single-tenant). ``role_auth_setup`` enables # workspaces; override it back off for this case. monkeypatch.setenv(MLFLOW_ENABLE_WORKSPACES.name, "false") role_auth_setup["login_as"]("outsider") with auth_module.app.test_request_context("/api/2.0/mlflow/users/list", method="GET"): assert auth_module.validate_can_list_users() is True def test_list_users_handler_eager_loads_scoped_roles(role_auth_setup): # Workspace admin: bulk response includes per-user roles, scoped to # workspaces the requester administers (plus self, unscoped). role_auth_setup["login_as"]("ws_admin_foo") with auth_module.app.test_request_context("/api/2.0/mlflow/users/list", method="GET"): response = auth_module.list_users() by_username = {u["username"]: u for u in response.get_json()["users"]} # Self: own admin role visible. assert {(r["workspace"], r["name"]) for r in by_username["ws_admin_foo"]["roles"]} == { ("foo", "admin-foo") } # Cross-user in foo: filtered to foo only (member-foo lives in foo). assert {(r["workspace"], r["name"]) for r in by_username["ws_member_foo"]["roles"]} == { ("foo", "member-foo") } # Cross-user outside requester's admin set: roles hidden. assert by_username["ws_admin_bar"]["roles"] == [] assert by_username["outsider"]["roles"] == [] def test_list_users_handler_super_admin_sees_every_role(role_auth_setup): role_auth_setup["login_as"]("super_admin") with auth_module.app.test_request_context("/api/2.0/mlflow/users/list", method="GET"): response = auth_module.list_users() by_username = {u["username"]: u for u in response.get_json()["users"]} assert {(r["workspace"], r["name"]) for r in by_username["ws_admin_foo"]["roles"]} == { ("foo", "admin-foo") } assert {(r["workspace"], r["name"]) for r in by_username["ws_admin_bar"]["roles"]} == { ("bar", "admin-bar") } assert {(r["workspace"], r["name"]) for r in by_username["ws_member_foo"]["roles"]} == { ("foo", "member-foo") } @pytest.mark.parametrize( ("actor", "expected"), [ ("ws_admin_foo", True), ("ws_admin_bar", True), ("ws_member_foo", False), ("outsider", False), ], ) def test_validate_can_create_user(role_auth_setup, actor, expected): role_auth_setup["login_as"](actor) with auth_module.app.test_request_context("/api/2.0/mlflow/users/create", method="POST"): assert auth_module.validate_can_create_user() is expected def test_validate_can_delete_user_stays_super_admin_only(role_auth_setup): # Regression: the create-user widening must not have leaked into delete. for actor in ("ws_admin_foo", "outsider"): role_auth_setup["login_as"](actor) with auth_module.app.test_request_context("/api/2.0/mlflow/users/delete", method="DELETE"): assert auth_module.validate_can_delete_user() is False def test_validate_can_view_user_roles_self_always_allowed(role_auth_setup): # A user can always read their own role list, even one with no roles. # Using ``outsider`` (zero roles) exercises the self-short-circuit without # any membership helping. role_auth_setup["login_as"]("outsider") with auth_module.app.test_request_context( "/api/3.0/mlflow/users/roles/list", method="GET", query_string={"username": "outsider"}, ): assert auth_module.validate_can_view_user_roles() is True @pytest.mark.parametrize( ("requester", "target", "expected"), [ ("super_admin", "ws_member_foo", True), ("ws_admin_foo", "ws_member_foo", True), ("ws_admin_bar", "ws_member_foo", False), ("ws_member_foo", "ws_admin_foo", False), ("outsider", "ws_member_foo", False), ], ) def test_validate_can_view_user_roles_cross_user(role_auth_setup, requester, target, expected): role_auth_setup["login_as"](requester) with auth_module.app.test_request_context( "/api/3.0/mlflow/users/roles/list", method="GET", query_string={"username": target}, ): assert auth_module.validate_can_view_user_roles() is expected def test_validate_can_view_user_roles_nonexistent_target_denied_for_non_admin( role_auth_setup, ): # Non-existent target: return False rather than leaking existence via the # RESOURCE_DOES_NOT_EXIST the handler would raise downstream. role_auth_setup["login_as"]("ws_admin_foo") with auth_module.app.test_request_context( "/api/3.0/mlflow/users/roles/list", method="GET", query_string={"username": "ghost"}, ): assert auth_module.validate_can_view_user_roles() is False def test_validate_can_view_user_roles_nonexistent_target_allowed_for_super_admin( role_auth_setup, ): # Super admin short-circuits before the target lookup — they're authorized # regardless of whether the target exists (the handler then 404s cleanly). role_auth_setup["login_as"]("super_admin") with auth_module.app.test_request_context( "/api/3.0/mlflow/users/roles/list", method="GET", query_string={"username": "ghost"}, ): assert auth_module.validate_can_view_user_roles() is True @pytest.mark.parametrize("shape", ["role_id", "role_permission_id"]) def test_validate_can_manage_roles_nonexistent_resource_denied(role_auth_setup, shape): # A non-admin pointing at a role/role_permission that doesn't exist fails # closed: _get_role_workspace_from_request returns None and the validator # treats that as unauthorized rather than leaking existence. role_auth_setup["login_as"]("ws_admin_foo") bogus_id = 999_999 if shape == "role_id": ctx = auth_module.app.test_request_context( "/api/3.0/mlflow/roles/get", method="GET", query_string={"role_id": str(bogus_id)}, ) else: ctx = auth_module.app.test_request_context( "/api/3.0/mlflow/roles/permissions/update", method="PATCH", json={"role_permission_id": bogus_id, "permission": READ.name}, ) with ctx: assert auth_module.validate_can_manage_roles() is False def test_validate_can_manage_roles_nonexistent_role_id_bypassed_by_super_admin( role_auth_setup, ): # Super admins skip the workspace resolution entirely — an unresolvable # role_id still produces True at the validator layer. role_auth_setup["login_as"]("super_admin") with auth_module.app.test_request_context( "/api/3.0/mlflow/roles/get", method="GET", query_string={"role_id": "999999"}, ): assert auth_module.validate_can_manage_roles() is True def test_validate_can_manage_roles_missing_workspace_params_raises(role_auth_setup): # No role_id / role_permission_id / workspace in the request body: the # resolver raises INVALID_PARAMETER_VALUE — callers that hit this path have # a client bug, and we surface it instead of silently denying. role_auth_setup["login_as"]("ws_admin_foo") with auth_module.app.test_request_context( "/api/3.0/mlflow/roles/create", method="POST", json={} ): with pytest.raises(MlflowException, match="must include one of"): auth_module.validate_can_manage_roles() def test_validate_can_manage_roles_blank_workspace_raises(role_auth_setup): role_auth_setup["login_as"]("ws_admin_foo") with auth_module.app.test_request_context( "/api/3.0/mlflow/roles/create", method="POST", json={"name": "new-role", "workspace": " "}, ): with pytest.raises(MlflowException, match="non-empty string"): auth_module.validate_can_manage_roles() def test_validate_can_manage_roles_propagates_param_coercion_errors(role_auth_setup): # Integration check: a non-integer role_id in the request surfaces the # coercion error through the validator chain rather than silently denying. role_auth_setup["login_as"]("ws_admin_foo") with auth_module.app.test_request_context( "/api/3.0/mlflow/roles/get", method="GET", query_string={"role_id": "not-an-int"}, ): with pytest.raises(MlflowException, match="must be an integer"): auth_module.validate_can_manage_roles() def test_role_permission_resolver_honors_default_workspace_autogrant(monkeypatch): """Resource-level resolution must fall back to ``default_permission`` for an ungranted user in the configured default workspace when ``grant_default_workspace_access=true``. Without this, deployments that relied on the implicit auto-grant pre-simplification suddenly see ``NO_PERMISSIONS`` for resources in the default workspace. """ monkeypatch.setenv(MLFLOW_ENABLE_WORKSPACES.name, "true") monkeypatch.setattr( auth_module, "auth_config", auth_module.auth_config._replace( default_permission=READ.name, grant_default_workspace_access=True, ), raising=False, ) default_workspace = "team-default" monkeypatch.setattr(auth_module, "_get_workspace_store", lambda: None, raising=False) monkeypatch.setattr( auth_module, "get_default_workspace_optional", lambda *args, **kwargs: (SimpleNamespace(name=default_workspace), True), raising=False, ) class DummyStore: def get_user(self, username): return SimpleNamespace(id=42, username=username) def get_role_permission_for_resource(self, *args, **kwargs): return None monkeypatch.setattr(auth_module, "store", DummyStore(), raising=False) monkeypatch.setattr( auth_module, "_get_resource_workspace", lambda *args, **kwargs: default_workspace, ) role_perm = auth_module._role_permission_for( username="alice", resource_type="experiment", resource_key="exp-1", workspace_lookup_id="exp-1", workspace_fetcher=lambda _id: SimpleNamespace(), workspace_label="experiment", ) perm = auth_module._get_role_permission_or_default(role_perm) assert perm.name == READ.name def test_role_permission_resolver_denies_in_non_default_workspace(monkeypatch): """The auto-grant only applies to the configured default workspace. An ungranted user in any other workspace must still get ``NO_PERMISSIONS``, even if ``grant_default_workspace_access`` is enabled. """ monkeypatch.setenv(MLFLOW_ENABLE_WORKSPACES.name, "true") monkeypatch.setattr( auth_module, "auth_config", auth_module.auth_config._replace( default_permission=READ.name, grant_default_workspace_access=True, ), raising=False, ) monkeypatch.setattr(auth_module, "_get_workspace_store", lambda: None, raising=False) monkeypatch.setattr( auth_module, "get_default_workspace_optional", lambda *args, **kwargs: (SimpleNamespace(name="team-default"), True), raising=False, ) class DummyStore: def get_user(self, username): return SimpleNamespace(id=42, username=username) def get_role_permission_for_resource(self, *args, **kwargs): return None monkeypatch.setattr(auth_module, "store", DummyStore(), raising=False) monkeypatch.setattr( auth_module, "_get_resource_workspace", lambda *args, **kwargs: "other-workspace", ) role_perm = auth_module._role_permission_for( username="alice", resource_type="experiment", resource_key="exp-1", workspace_lookup_id="exp-1", workspace_fetcher=lambda _id: SimpleNamespace(), workspace_label="experiment", ) perm = auth_module._get_role_permission_or_default(role_perm) assert perm.name == NO_PERMISSIONS.name def test_list_user_role_permissions_workspace_is_default_when_workspaces_disabled( tmp_path, monkeypatch ): # ``_UserRolePermissionRow.workspace`` is typed ``str``, not ``str | None``. # When workspaces are disabled, ``_get_active_workspace_name()`` returns the # string ``"default"`` (``DEFAULT_WORKSPACE_NAME``) and every role write # threads that through ``SqlRole.workspace`` (``nullable=False``). Pins # that LIST surfaces ``"default"`` rather than ``None`` in this mode. from mlflow.utils.workspace_utils import DEFAULT_WORKSPACE_NAME monkeypatch.delenv(MLFLOW_ENABLE_WORKSPACES.name, raising=False) db_uri = f"sqlite:///{tmp_path / 'auth-store.db'}" auth_store = SqlAlchemyStore() auth_store.init_db(db_uri) monkeypatch.setattr(auth_module, "store", auth_store, raising=False) username = "alice" auth_store.create_user(username, "supersecurepassword", is_admin=False) auth_store.grant_user_resource_permission(username, "experiment", "exp-1", READ.name) is_admin, rows = auth_module._list_user_role_permissions(username) assert is_admin is False assert len(rows) == 1 row = rows[0] assert isinstance(row.workspace, str) assert row.workspace == DEFAULT_WORKSPACE_NAME assert row.resource_type == "experiment" assert row.resource_pattern == "exp-1" assert row.permission == READ.name # Direct grants land on the synthetic ``__user___`` role. assert row.role_name == f"__user_{auth_store.get_user(username).id}__" def test_list_user_role_permissions_aggregates_synthetic_and_custom_roles(tmp_path, monkeypatch): # Direct grants flow into the synthetic ``__user___`` role; a workspace # admin can also assign the user to a named, hand-authored role. LIST must # surface BOTH so the FE can split the "Direct permissions" tab (synthetic # rows) from the "Role permissions" tab (everything else) by ``role_name``. monkeypatch.delenv(MLFLOW_ENABLE_WORKSPACES.name, raising=False) db_uri = f"sqlite:///{tmp_path / 'auth-store.db'}" auth_store = SqlAlchemyStore() auth_store.init_db(db_uri) monkeypatch.setattr(auth_module, "store", auth_store, raising=False) username = "alice" user = auth_store.create_user(username, "supersecurepassword", is_admin=False) auth_store.grant_user_resource_permission(username, "experiment", "exp-1", READ.name) custom_role = auth_store.create_role(name="ml-reviewer", workspace="default") auth_store.add_role_permission(custom_role.id, "registered_model", "*", READ.name) auth_store.assign_role_to_user(user.id, custom_role.id) is_admin, rows = auth_module._list_user_role_permissions(username) assert is_admin is False assert len(rows) == 2 by_role = {r.role_name: r for r in rows} synthetic_name = f"__user_{user.id}__" assert synthetic_name in by_role assert "ml-reviewer" in by_role syn_row = by_role[synthetic_name] assert syn_row.resource_type == "experiment" assert syn_row.resource_pattern == "exp-1" assert syn_row.permission == READ.name custom_row = by_role["ml-reviewer"] assert custom_row.role_id == custom_role.id assert custom_row.resource_type == "registered_model" assert custom_row.resource_pattern == "*" assert custom_row.permission == READ.name # Distinct rows really do come from distinct roles (no aliasing on role_id). assert syn_row.role_id != custom_row.role_id def test_list_user_role_permissions_empty_for_user_with_no_roles(tmp_path, monkeypatch): # A freshly-created non-admin user with no grants must surface as # ``(is_admin=False, rows=[])`` — not raise, and not leak rows from other users. monkeypatch.delenv(MLFLOW_ENABLE_WORKSPACES.name, raising=False) db_uri = f"sqlite:///{tmp_path / 'auth-store.db'}" auth_store = SqlAlchemyStore() auth_store.init_db(db_uri) monkeypatch.setattr(auth_module, "store", auth_store, raising=False) auth_store.create_user("alice", "supersecurepassword", is_admin=False) # Second user with a grant — must not leak into alice's rows. auth_store.create_user("bob", "supersecurepassword", is_admin=False) auth_store.grant_user_resource_permission("bob", "experiment", "exp-1", READ.name) is_admin, rows = auth_module._list_user_role_permissions("alice") assert is_admin is False assert rows == [] def test_list_user_role_permissions_admin_flag_propagates(tmp_path, monkeypatch): # ``is_admin`` comes from the User row, not from any grants — admins can # have zero permission rows and still surface ``is_admin=True``. The FE # relies on this to skip permission-fetch entirely for super admins. monkeypatch.delenv(MLFLOW_ENABLE_WORKSPACES.name, raising=False) db_uri = f"sqlite:///{tmp_path / 'auth-store.db'}" auth_store = SqlAlchemyStore() auth_store.init_db(db_uri) monkeypatch.setattr(auth_module, "store", auth_store, raising=False) auth_store.create_user("rootuser", "supersecurepassword", is_admin=True) is_admin, rows = auth_module._list_user_role_permissions("rootuser") assert is_admin is True assert rows == [] def test_list_user_role_permissions_multiple_perms_on_one_role(tmp_path, monkeypatch): # One role with two RolePermission rows must produce two LIST rows that # share role_id / role_name / workspace but differ on resource_type / # permission. Pins the flatten-on-role.permissions behavior. monkeypatch.delenv(MLFLOW_ENABLE_WORKSPACES.name, raising=False) db_uri = f"sqlite:///{tmp_path / 'auth-store.db'}" auth_store = SqlAlchemyStore() auth_store.init_db(db_uri) monkeypatch.setattr(auth_module, "store", auth_store, raising=False) user = auth_store.create_user("alice", "supersecurepassword", is_admin=False) role = auth_store.create_role(name="ml-power-user", workspace="default") auth_store.add_role_permission(role.id, "experiment", "*", READ.name) auth_store.add_role_permission(role.id, "registered_model", "*", MANAGE.name) auth_store.assign_role_to_user(user.id, role.id) is_admin, rows = auth_module._list_user_role_permissions("alice") assert is_admin is False assert len(rows) == 2 assert {(r.resource_type, r.permission) for r in rows} == { ("experiment", READ.name), ("registered_model", MANAGE.name), } assert {r.role_id for r in rows} == {role.id} assert {r.role_name for r in rows} == {"ml-power-user"} def test_list_user_role_permissions_workspace_reflects_role_workspace(tmp_path, monkeypatch): # With workspaces enabled, ``row.workspace`` is the *role's* workspace — # i.e., the workspace where the resource lives — not always # ``DEFAULT_WORKSPACE_NAME``. Pin against silent collapsing of the workspace # field for non-default workspaces. monkeypatch.setenv(MLFLOW_ENABLE_WORKSPACES.name, "true") db_uri = f"sqlite:///{tmp_path / 'auth-store.db'}" auth_store = SqlAlchemyStore() auth_store.init_db(db_uri) monkeypatch.setattr(auth_module, "store", auth_store, raising=False) user = auth_store.create_user("alice", "supersecurepassword", is_admin=False) role_a = auth_store.create_role(name="viewer", workspace="team-a") auth_store.add_role_permission(role_a.id, "experiment", "*", READ.name) auth_store.assign_role_to_user(user.id, role_a.id) role_b = auth_store.create_role(name="viewer", workspace="team-b") auth_store.add_role_permission(role_b.id, "experiment", "*", READ.name) auth_store.assign_role_to_user(user.id, role_b.id) _, rows = auth_module._list_user_role_permissions("alice") assert {r.workspace for r in rows} == {"team-a", "team-b"} def _list_user_permissions_response(monkeypatch, requester: str, target: str) -> dict[str, object]: """Invoke ``list_user_permissions`` as ``requester`` for ``target`` and return JSON.""" monkeypatch.setattr( auth_module, "authenticate_request", lambda: SimpleNamespace(username=requester) ) with auth_module.app.test_request_context( "/api/3.0/mlflow/users/permissions/list", method="GET", query_string={"username": target}, ): response = auth_module.list_user_permissions() return json.loads(response.get_data(as_text=True)) def test_list_user_permissions_admin_sees_rows_across_all_workspaces(tmp_path, monkeypatch): # Platform admins must see EVERY role-derived row for the target user, # regardless of which workspace the row sits in. This is the path the # admin UI's "User detail" page hits when an admin clicks any user. monkeypatch.setenv(MLFLOW_ENABLE_WORKSPACES.name, "true") db_uri = f"sqlite:///{tmp_path / 'auth-store.db'}" auth_store = SqlAlchemyStore() auth_store.init_db(db_uri) monkeypatch.setattr(auth_module, "store", auth_store, raising=False) auth_store.create_user("root", "supersecurepassword", is_admin=True) target = auth_store.create_user("alice", "supersecurepassword", is_admin=False) role_a = auth_store.create_role(name="viewer", workspace="team-a") auth_store.add_role_permission(role_a.id, "experiment", "*", READ.name) auth_store.assign_role_to_user(target.id, role_a.id) role_b = auth_store.create_role(name="viewer", workspace="team-b") auth_store.add_role_permission(role_b.id, "experiment", "*", READ.name) auth_store.assign_role_to_user(target.id, role_b.id) payload = _list_user_permissions_response(monkeypatch, requester="root", target="alice") assert payload["is_admin"] is False # is_admin reflects the *target*, not the caller. assert {p["workspace"] for p in payload["permissions"]} == {"team-a", "team-b"} def test_list_user_permissions_self_sees_all_rows(tmp_path, monkeypatch): # Self-view sees everything the user holds, across every workspace, with no # admin-workspace filtering. ``is_admin`` here mirrors the caller because # caller == target. monkeypatch.setenv(MLFLOW_ENABLE_WORKSPACES.name, "true") db_uri = f"sqlite:///{tmp_path / 'auth-store.db'}" auth_store = SqlAlchemyStore() auth_store.init_db(db_uri) monkeypatch.setattr(auth_module, "store", auth_store, raising=False) alice = auth_store.create_user("alice", "supersecurepassword", is_admin=False) role_a = auth_store.create_role(name="viewer", workspace="team-a") auth_store.add_role_permission(role_a.id, "experiment", "*", READ.name) auth_store.assign_role_to_user(alice.id, role_a.id) role_b = auth_store.create_role(name="viewer", workspace="team-b") auth_store.add_role_permission(role_b.id, "experiment", "*", READ.name) auth_store.assign_role_to_user(alice.id, role_b.id) payload = _list_user_permissions_response(monkeypatch, requester="alice", target="alice") assert payload["is_admin"] is False assert {p["workspace"] for p in payload["permissions"]} == {"team-a", "team-b"} def test_list_user_permissions_workspace_admin_scoped_to_own_workspaces(tmp_path, monkeypatch): # A workspace admin of team-a viewing alice (who has rows in team-a + team-b) # must see ONLY team-a rows. The team-b row would leak cross-workspace state # the requester has no claim over. Security regression guard. monkeypatch.setenv(MLFLOW_ENABLE_WORKSPACES.name, "true") db_uri = f"sqlite:///{tmp_path / 'auth-store.db'}" auth_store = SqlAlchemyStore() auth_store.init_db(db_uri) monkeypatch.setattr(auth_module, "store", auth_store, raising=False) auth_store.create_user("wp-admin", "supersecurepassword", is_admin=False) auth_store.set_workspace_permission("team-a", "wp-admin", MANAGE.name) alice = auth_store.create_user("alice", "supersecurepassword", is_admin=False) role_a = auth_store.create_role(name="viewer", workspace="team-a") auth_store.add_role_permission(role_a.id, "experiment", "*", READ.name) auth_store.assign_role_to_user(alice.id, role_a.id) role_b = auth_store.create_role(name="viewer", workspace="team-b") auth_store.add_role_permission(role_b.id, "experiment", "*", READ.name) auth_store.assign_role_to_user(alice.id, role_b.id) payload = _list_user_permissions_response(monkeypatch, requester="wp-admin", target="alice") workspaces = {p["workspace"] for p in payload["permissions"]} assert workspaces == {"team-a"}, workspaces def test_list_current_user_permissions_returns_caller_rows_and_admin_flag(tmp_path, monkeypatch): # ``/users/current/permissions`` is the unauthenticated-safe self path the # FE uses on bootstrap to decide which nav items to render. Caller == target # implicitly; ``is_admin`` reflects the caller; rows cover every workspace. monkeypatch.setenv(MLFLOW_ENABLE_WORKSPACES.name, "true") db_uri = f"sqlite:///{tmp_path / 'auth-store.db'}" auth_store = SqlAlchemyStore() auth_store.init_db(db_uri) monkeypatch.setattr(auth_module, "store", auth_store, raising=False) alice = auth_store.create_user("alice", "supersecurepassword", is_admin=False) role = auth_store.create_role(name="viewer", workspace="team-a") auth_store.add_role_permission(role.id, "experiment", "*", READ.name) auth_store.assign_role_to_user(alice.id, role.id) monkeypatch.setattr( auth_module, "authenticate_request", lambda: SimpleNamespace(username="alice") ) with auth_module.app.test_request_context( "/api/3.0/mlflow/users/current/permissions", method="GET" ): response = auth_module.list_current_user_permissions() payload = json.loads(response.get_data(as_text=True)) assert payload["is_admin"] is False assert len(payload["permissions"]) == 1 assert payload["permissions"][0]["workspace"] == "team-a" assert payload["permissions"][0]["role_name"] == "viewer" def test_default_permission_floors_lesser_role_grant(monkeypatch): monkeypatch.setattr( auth_module, "auth_config", auth_module.auth_config._replace(default_permission=EDIT.name), raising=False, ) perm = auth_module._get_role_permission_or_default(lambda: READ) assert perm.name == EDIT.name def test_default_permission_does_not_downgrade_higher_role_grant(monkeypatch): monkeypatch.setattr( auth_module, "auth_config", auth_module.auth_config._replace(default_permission=READ.name), raising=False, ) perm = auth_module._get_role_permission_or_default(lambda: MANAGE) assert perm.name == MANAGE.name def test_default_permission_does_not_override_explicit_no_permissions(monkeypatch): # NO_PERMISSIONS is the explicit-deny carve-out — must survive the floor. monkeypatch.setattr( auth_module, "auth_config", auth_module.auth_config._replace(default_permission=MANAGE.name), raising=False, ) perm = auth_module._get_role_permission_or_default(lambda: NO_PERMISSIONS) assert perm.name == NO_PERMISSIONS.name def test_default_permission_kicks_in_when_no_grant_matches(monkeypatch): # None (no grant matched, workspaces disabled) → fall through to default. monkeypatch.setattr( auth_module, "auth_config", auth_module.auth_config._replace(default_permission=EDIT.name), raising=False, ) perm = auth_module._get_role_permission_or_default(lambda: None) assert perm.name == EDIT.name def test_user_can_create_in_default_workspace_via_autogrant(monkeypatch): """``_user_can_create_in_workspace`` must honor ``grant_default_workspace_access`` so an ungranted user in the default workspace can still create when ``default_permission.can_use`` is true. Regression guard for the legacy-endpoint simplification, which dropped the auto-grant fallback when ``_workspace_permission`` was retired. """ monkeypatch.setenv(MLFLOW_ENABLE_WORKSPACES.name, "true") auth = SimpleNamespace(username="alice") monkeypatch.setattr(auth_module, "authenticate_request", lambda: auth) monkeypatch.setattr( auth_module, "auth_config", auth_module.auth_config._replace( default_permission=USE.name, grant_default_workspace_access=True, ), raising=False, ) default_workspace = "team-default" monkeypatch.setattr(auth_module, "_get_workspace_store", lambda: None, raising=False) monkeypatch.setattr( auth_module, "get_default_workspace_optional", lambda *args, **kwargs: (SimpleNamespace(name=default_workspace), True), raising=False, ) class DummyStore: def get_user(self, username): return SimpleNamespace(id=42, username=username) def get_role_permission_for_resource(self, *args, **kwargs): return None monkeypatch.setattr(auth_module, "store", DummyStore(), raising=False) # Default workspace + autogrant + USE → allowed. with workspace_context.WorkspaceContext(default_workspace): assert auth_module._user_can_create_in_workspace() # Same config but a non-default workspace → still denied. with workspace_context.WorkspaceContext("team-other"): assert not auth_module._user_can_create_in_workspace() def test_user_cannot_create_via_autogrant_when_default_permission_lacks_use(monkeypatch): """The auto-grant create-gate gates on ``default_permission.can_use``. If the operator pinned ``default_permission=READ`` (read-only access), the fallback must not allow create. """ monkeypatch.setenv(MLFLOW_ENABLE_WORKSPACES.name, "true") auth = SimpleNamespace(username="alice") monkeypatch.setattr(auth_module, "authenticate_request", lambda: auth) monkeypatch.setattr( auth_module, "auth_config", auth_module.auth_config._replace( default_permission=READ.name, grant_default_workspace_access=True, ), raising=False, ) default_workspace = "team-default" monkeypatch.setattr(auth_module, "_get_workspace_store", lambda: None, raising=False) monkeypatch.setattr( auth_module, "get_default_workspace_optional", lambda *args, **kwargs: (SimpleNamespace(name=default_workspace), True), raising=False, ) class DummyStore: def get_user(self, username): return SimpleNamespace(id=42, username=username) def get_role_permission_for_resource(self, *args, **kwargs): return None monkeypatch.setattr(auth_module, "store", DummyStore(), raising=False) with workspace_context.WorkspaceContext(default_workspace): assert not auth_module._user_can_create_in_workspace() def test_role_based_read_predicate_ignores_no_permissions_grants(monkeypatch): monkeypatch.delenv(MLFLOW_ENABLE_WORKSPACES.name, raising=False) monkeypatch.setattr( auth_module, "auth_config", auth_module.auth_config._replace(default_permission=READ.name), raising=False, ) class DummyStore: def get_user(self, username): return SimpleNamespace(id=42, username=username) def list_role_grants_for_user_in_workspace(self, *args, **kwargs): return [ ("*", NO_PERMISSIONS.name), ("exp-allowed", READ.name), ("exp-explicit-deny", NO_PERMISSIONS.name), ] monkeypatch.setattr(auth_module, "store", DummyStore(), raising=False) predicate = auth_module._role_based_read_predicate("alice", "experiment") # Specific positive grant wins. assert predicate("exp-allowed") # NO_PERMISSIONS wildcard is ignored; default READ fallback applies. assert predicate("exp-other") # Per-resource NO_PERMISSIONS is ignored; default READ fallback applies. assert predicate("exp-explicit-deny") # ============================================================================= # Unified per-user permission convenience APIs — validator dispatcher tests # ============================================================================= def _scorer_resource_id(experiment_id: str, scorer_name: str) -> str: from mlflow.server.auth.sqlalchemy_store import SqlAlchemyStore return SqlAlchemyStore._scorer_pattern(experiment_id, scorer_name) @pytest.mark.parametrize( ("resource_type", "resource_id"), [ ("experiment", "exp-1"), ("registered_model", "model-xyz"), ("scorer", "exp-1/score-1"), ("gateway_secret", "secret-1"), ("gateway_endpoint", "endpoint-1"), ("gateway_model_definition", "model-def-1"), ], ) def test_validate_can_manage_resource_workspace_manage_allows( workspace_permission_setup, resource_type, resource_id ): # Workspace MANAGE (the fixture's default) grants per-resource MANAGE on every # resource type in the workspace — the dispatcher must route the request # through the same code path the legacy validators use. store = workspace_permission_setup["store"] username = workspace_permission_setup["username"] _set_workspace_permission(store, username, MANAGE.name) with auth_module.app.test_request_context( "/api/3.0/mlflow/users/permissions/grant", method="POST", json={ "username": username, "resource_type": resource_type, "resource_id": resource_id, "permission": "READ", }, ): assert auth_module.validate_can_manage_resource() def test_validate_can_manage_resource_per_resource_manage_allows( workspace_permission_setup, ): # Per-resource MANAGE delegation: a user with MANAGE on exp-1 (and nothing # else) can grant other users access to exp-1, mirroring the legacy # ``validate_can_manage_experiment`` semantics. store = workspace_permission_setup["store"] username = workspace_permission_setup["username"] user_id = store.get_user(username).id _set_workspace_permission(store, username, NO_PERMISSIONS.name) role = store.create_role(name="exp-1-manager", workspace="team-a") store.add_role_permission(role.id, "experiment", "exp-1", MANAGE.name) store.assign_role_to_user(user_id, role.id) with auth_module.app.test_request_context( "/api/3.0/mlflow/users/permissions/grant", method="POST", json={ "username": username, "resource_type": "experiment", "resource_id": "exp-1", "permission": "READ", }, ): assert auth_module.validate_can_manage_resource() def test_validate_can_manage_resource_other_resource_denied(workspace_permission_setup): # MANAGE on exp-1 doesn't extend to exp-2. store = workspace_permission_setup["store"] username = workspace_permission_setup["username"] user_id = store.get_user(username).id _set_workspace_permission(store, username, NO_PERMISSIONS.name) role = store.create_role(name="exp-1-manager", workspace="team-a") store.add_role_permission(role.id, "experiment", "exp-1", MANAGE.name) store.assign_role_to_user(user_id, role.id) with auth_module.app.test_request_context( "/api/3.0/mlflow/users/permissions/grant", method="POST", json={ "username": username, "resource_type": "experiment", "resource_id": "exp-2", "permission": "READ", }, ): assert not auth_module.validate_can_manage_resource() def test_validate_can_manage_resource_no_grant_denied(workspace_permission_setup): store = workspace_permission_setup["store"] username = workspace_permission_setup["username"] _set_workspace_permission(store, username, NO_PERMISSIONS.name) with auth_module.app.test_request_context( "/api/3.0/mlflow/users/permissions/grant", method="POST", json={ "username": username, "resource_type": "experiment", "resource_id": "exp-1", "permission": "READ", }, ): assert not auth_module.validate_can_manage_resource() def test_validate_can_manage_resource_workspace_use_insufficient(workspace_permission_setup): # Workspace USE (regular member) doesn't grant per-resource MANAGE — only # the resource owner's grant tier does. store = workspace_permission_setup["store"] username = workspace_permission_setup["username"] _set_workspace_permission(store, username, USE.name) with auth_module.app.test_request_context( "/api/3.0/mlflow/users/permissions/grant", method="POST", json={ "username": username, "resource_type": "experiment", "resource_id": "exp-1", "permission": "READ", }, ): assert not auth_module.validate_can_manage_resource() def test_validate_can_manage_resource_scorer_dispatch(workspace_permission_setup): # The scorer ``resource_id`` is the compound pattern # ``"/"``. The dispatcher must split # off the experiment_id prefix for workspace resolution. store = workspace_permission_setup["store"] username = workspace_permission_setup["username"] _set_workspace_permission(store, username, MANAGE.name) pattern = _scorer_resource_id("exp-1", "score-1") with auth_module.app.test_request_context( "/api/3.0/mlflow/users/permissions/grant", method="POST", json={ "username": username, "resource_type": "scorer", "resource_id": pattern, "permission": "READ", }, ): assert auth_module.validate_can_manage_resource() def test_validate_can_manage_resource_scorer_missing_delimiter_raises(workspace_permission_setup): # ``resource_id`` without the ``/`` delimiter cannot be split into # ``experiment_id`` for workspace resolution. store = workspace_permission_setup["store"] username = workspace_permission_setup["username"] _set_workspace_permission(store, username, MANAGE.name) with auth_module.app.test_request_context( "/api/3.0/mlflow/users/permissions/grant", method="POST", json={ "username": username, "resource_type": "scorer", "resource_id": "missing-delimiter", "permission": "READ", }, ): with pytest.raises(MlflowException, match="Expected '/'"): auth_module.validate_can_manage_resource() def test_validate_can_manage_resource_workspace_resource_type_rejected( workspace_permission_setup, ): # ``workspace`` is intentionally excluded from the unified API — workspace # grants live behind set_workspace_permission / delete_workspace_permission. store = workspace_permission_setup["store"] username = workspace_permission_setup["username"] _set_workspace_permission(store, username, MANAGE.name) with auth_module.app.test_request_context( "/api/3.0/mlflow/users/permissions/grant", method="POST", json={ "username": username, "resource_type": "workspace", "resource_id": "*", "permission": "MANAGE", }, ): with pytest.raises(MlflowException, match="is not supported by the per-user"): auth_module.validate_can_manage_resource() def test_validate_can_get_user_permission_self_check_allowed( workspace_permission_setup, ): # A non-admin user can always check their own permissions, even without # any workspace presence. store = workspace_permission_setup["store"] username = workspace_permission_setup["username"] _set_workspace_permission(store, username, NO_PERMISSIONS.name) with auth_module.app.test_request_context( "/api/3.0/mlflow/users/permissions/get", method="GET", query_string={ "username": username, "resource_type": "experiment", "resource_id": "exp-1", }, ): assert auth_module.validate_can_get_user_permission() def test_validate_can_get_user_permission_admin_short_circuits( workspace_permission_setup, monkeypatch ): # Platform admins bypass the workspace check. store = workspace_permission_setup["store"] admin_username = "platform-admin" store.create_user(admin_username, "supersecurepassword", is_admin=True) monkeypatch.setattr( auth_module, "authenticate_request", lambda: SimpleNamespace(username=admin_username), ) target = workspace_permission_setup["username"] with auth_module.app.test_request_context( "/api/3.0/mlflow/users/permissions/get", method="GET", query_string={ "username": target, "resource_type": "experiment", "resource_id": "exp-1", }, ): assert auth_module.validate_can_get_user_permission() def test_validate_can_get_user_permission_admin_probes_other_workspace( workspace_permission_setup, monkeypatch ): # Platform admins bypass the workspace check globally — even for resources in # workspaces they have no role in. Pins the is_admin short-circuit on the # cross-workspace path that ``_cross_workspace_probe_denied`` denies for non-admins. store = workspace_permission_setup["store"] admin_username = "platform-admin" store.create_user(admin_username, "supersecurepassword", is_admin=True) monkeypatch.setattr( auth_module, "authenticate_request", lambda: SimpleNamespace(username=admin_username), ) target = workspace_permission_setup["username"] # team-b experiment — admin has no membership in team-b. auth_module._get_tracking_store()._experiment_workspaces["exp-team-b"] = "team-b" with auth_module.app.test_request_context( "/api/3.0/mlflow/users/permissions/get", method="GET", query_string={ "username": target, "resource_type": "experiment", "resource_id": "exp-team-b", }, ): assert auth_module.validate_can_get_user_permission() def test_validate_can_get_user_permission_cross_user_requires_admin( workspace_permission_setup, ): # A non-admin requester without workspace MANAGE in the resource's workspace # cannot check another user's permissions. ``alice`` holds USE in team-a (not # MANAGE), so the workspace-admin gate fails even though the resource lives # in alice's workspace. store = workspace_permission_setup["store"] requester = workspace_permission_setup["username"] target = "bob" store.create_user(target, "supersecurepassword", is_admin=False) _set_workspace_permission(store, requester, USE.name) with auth_module.app.test_request_context( "/api/3.0/mlflow/users/permissions/get", method="GET", query_string={ "username": target, "resource_type": "experiment", "resource_id": "exp-1", }, ): assert not auth_module.validate_can_get_user_permission() def test_validate_can_get_user_permission_wp_admin_scoped_to_resource_workspace( workspace_permission_setup, ): # A workspace admin in team-a can check another user's permissions on resources # in team-a. The scoping is by **resource workspace** (not by target-user # presence), so the target need not have a role in team-a for the gate to allow. store = workspace_permission_setup["store"] requester = workspace_permission_setup["username"] _set_workspace_permission(store, requester, MANAGE.name) target = "bob" store.create_user(target, "supersecurepassword", is_admin=False) with auth_module.app.test_request_context( "/api/3.0/mlflow/users/permissions/get", method="GET", query_string={ "username": target, "resource_type": "experiment", "resource_id": "exp-1", }, ): assert auth_module.validate_can_get_user_permission() def test_validate_can_get_user_permission_cross_workspace_probe_denied( workspace_permission_setup, ): # Security gate: a workspace admin of team-a must NOT be able to probe a # target user's permissions on a resource in team-b. Closes the # cross-workspace information-disclosure gap. store = workspace_permission_setup["store"] requester = workspace_permission_setup["username"] _set_workspace_permission(store, requester, MANAGE.name) target = "bob" store.create_user(target, "supersecurepassword", is_admin=False) auth_module._get_tracking_store()._experiment_workspaces["exp-team-b"] = "team-b" with auth_module.app.test_request_context( "/api/3.0/mlflow/users/permissions/get", method="GET", query_string={ "username": target, "resource_type": "experiment", "resource_id": "exp-team-b", }, ): assert not auth_module.validate_can_get_user_permission() def test_validate_can_get_user_permission_unknown_resource_denied( workspace_permission_setup, ): # If the resource can't be resolved to a workspace (e.g. it doesn't exist), # the gate must deny — otherwise a caller could probe across all workspaces # by using a non-existent ID. store = workspace_permission_setup["store"] requester = workspace_permission_setup["username"] _set_workspace_permission(store, requester, MANAGE.name) target = "bob" store.create_user(target, "supersecurepassword", is_admin=False) with auth_module.app.test_request_context( "/api/3.0/mlflow/users/permissions/get", method="GET", query_string={ "username": target, "resource_type": "experiment", "resource_id": "exp-does-not-exist", }, ): assert not auth_module.validate_can_get_user_permission()