Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions Dockerfile.allinone
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,8 @@ RUN apt-get update && apt-get install -y --no-install-recommends \
nginx \
supervisor \
gcc \
libc6-dev \
linux-libc-dev \
curl \
&& rm -rf /var/lib/apt/lists/*

Expand Down
60 changes: 55 additions & 5 deletions api/services/auth_services/authorization_service.py
Original file line number Diff line number Diff line change
Expand Up @@ -47,6 +47,40 @@ def endpoint_group_role_name(role_level: str) -> str:
return f"group:{ep_uuid}:{role_level}"


def endpoint_group_role_names(role_level: str) -> List[str]:
"""
Return accepted per-endpoint role names for ``role_level``.

The deployment's endpoint access boundary is configured through
``GROUP_NAMES``. Keycloak group roles are emitted as
``group:{group_path}:{role}``, so role checks must accept those group
paths directly. ``AFFINITIES_EP_UUID`` forms are still accepted for
older deployments that configured roles that way, but Affinities is
not required for the viewer/writer/admin model.
"""
role_names = [
f"group:{group_name}:{role_level}" for group_name in get_allowed_groups()
]

ep_uuid = (affinities_settings.ep_uuid or "").strip()
if not ep_uuid:
return list(dict.fromkeys(role_names))

normalized_ep = normalize_group_path(ep_uuid)
role_names.extend(
[
endpoint_group_role_name(role_level),
f"group:{normalized_ep}:{role_level}",
]
)
if "/" not in normalized_ep:
role_names.append(f"group:ndp_ep/{ep_uuid}:{role_level}")
if not normalized_ep.startswith("ep-"):
role_names.append(f"group:ndp_ep/ep-{ep_uuid}:{role_level}")

return list(dict.fromkeys(role_names))


def normalize_group_path(path: str) -> str:
"""
Normalize a group path for comparison.
Expand Down Expand Up @@ -372,6 +406,21 @@ def get_user_for_endpoint_access(
return user_info


def _normalize_role_name(role: str) -> str:
"""
Normalize a role name for comparison.

For Keycloak group roles, normalize the group path segment too, so
``group:/ndp_ep/ep-1:admin`` and ``group:ndp_ep/ep-1:admin`` compare
as the same role.
"""
value = role.strip().lower()
if value.startswith("group:") and ":" in value[len("group:") :]:
group_path, role_level = value[len("group:") :].rsplit(":", 1)
return f"group:{normalize_group_path(group_path)}:{role_level.strip()}"
return value


def _has_any_role(user_info: Dict[str, Any], candidates) -> bool:
"""
Return True if the user carries any of the role names in ``candidates``.
Expand All @@ -381,14 +430,15 @@ def _has_any_role(user_info: Dict[str, Any], candidates) -> bool:
Empty strings in ``candidates`` are skipped (callers may pass them
when AFFINITIES_EP_UUID is not configured).
"""
targets = {c.strip().lower() for c in candidates if c}
targets = {_normalize_role_name(c) for c in candidates if c}
if not targets:
return False
roles = user_info.get("roles", [])
if not isinstance(roles, list):
return False
return any(
isinstance(role, str) and role.strip().lower() in targets for role in roles
isinstance(role, str) and _normalize_role_name(role) in targets
for role in roles
)


Expand All @@ -406,7 +456,7 @@ def is_admin(user_info: Dict[str, Any]) -> bool:
user_info,
[
ADMIN_ROLE_NAME,
endpoint_group_role_name("admin"),
*endpoint_group_role_names("admin"),
endpoint_admin_role_name(),
],
)
Expand All @@ -421,7 +471,7 @@ def is_writer(user_info: Dict[str, Any]) -> bool:
return True
return _has_any_role(
user_info,
[WRITER_ROLE_NAME, endpoint_group_role_name("writer")],
[WRITER_ROLE_NAME, *endpoint_group_role_names("writer")],
)


Expand All @@ -434,7 +484,7 @@ def is_viewer(user_info: Dict[str, Any]) -> bool:
return True
return _has_any_role(
user_info,
[VIEWER_ROLE_NAME, endpoint_group_role_name("viewer")],
[VIEWER_ROLE_NAME, *endpoint_group_role_names("viewer")],
)


Expand Down
5 changes: 3 additions & 2 deletions api/services/auth_services/get_current_user.py
Original file line number Diff line number Diff line change
Expand Up @@ -40,10 +40,11 @@ def get_current_user(token_data=Depends(security)) -> Dict[str, Any]:
"""
token = token_data.credentials

# Handle test token
# Handle local test token. This bypasses the external identity provider
# but still returns the same role shape normal authorization expects.
if token == swagger_settings.test_token:
return {
"roles": ["admin", "user"],
"roles": ["ndp_admin"],
"groups": [],
"sub": "test_user",
"username": "Test User",
Expand Down
35 changes: 35 additions & 0 deletions tests/test_authorization_service.py
Original file line number Diff line number Diff line change
Expand Up @@ -684,6 +684,12 @@ def _patch_uuid(self, uuid):
ep_uuid=uuid,
)

def _patch_group_names(self, group_names):
return patch(
"api.services.auth_services.authorization_service.swagger_settings",
group_names=group_names,
)

EP = "11111111-2222-3333-4444-555555555555"

def test_is_admin_via_global_role(self):
Expand All @@ -694,6 +700,28 @@ def test_is_admin_via_canonical_per_ep_role(self):
user = {"roles": [f"group:{self.EP}:admin"]}
assert is_admin(user) is True

def test_is_admin_via_keycloak_group_path_role(self):
with self._patch_uuid(f"ep-{self.EP}"):
user = {"roles": [f"group:ndp_ep/ep-{self.EP}:admin"]}
assert is_admin(user) is True
assert effective_role(user) == "admin"

def test_is_admin_via_keycloak_group_path_role_with_bare_uuid_config(self):
with self._patch_uuid(self.EP):
user = {"roles": [f"group:ndp_ep/ep-{self.EP}:admin"]}
assert is_admin(user) is True

def test_is_admin_via_group_names_without_affinities_uuid(self):
with self._patch_uuid(""), self._patch_group_names(f"ndp_ep/ep-{self.EP}"):
user = {"roles": [f"group:ndp_ep/ep-{self.EP}:admin"]}
assert is_admin(user) is True
assert effective_role(user) == "admin"

def test_is_admin_normalizes_leading_slash_in_group_path_role(self):
with self._patch_uuid(f"ndp_ep/ep-{self.EP}"):
user = {"roles": [f"group:/ndp_ep/ep-{self.EP}:admin"]}
assert is_admin(user) is True

def test_is_admin_via_legacy_per_ep_role_is_still_supported(self):
with self._patch_uuid(self.EP):
user = {"roles": [f"{self.EP}_admin"]}
Expand All @@ -709,6 +737,7 @@ def test_is_writer_includes_writer_and_admin_but_not_viewer(self):
with self._patch_uuid(self.EP):
assert is_writer({"roles": [WRITER_ROLE_NAME]}) is True
assert is_writer({"roles": [f"group:{self.EP}:writer"]}) is True
assert is_writer({"roles": [f"group:ndp_ep/{self.EP}:writer"]}) is True
assert is_writer({"roles": [ADMIN_ROLE_NAME]}) is True
assert is_writer({"roles": [VIEWER_ROLE_NAME]}) is False
assert is_writer({"roles": [f"group:{self.EP}:viewer"]}) is False
Expand All @@ -718,10 +747,16 @@ def test_is_viewer_includes_every_tier_above_none(self):
with self._patch_uuid(self.EP):
assert is_viewer({"roles": [VIEWER_ROLE_NAME]}) is True
assert is_viewer({"roles": [f"group:{self.EP}:viewer"]}) is True
assert is_viewer({"roles": [f"group:ndp_ep/{self.EP}:viewer"]}) is True
assert is_viewer({"roles": [WRITER_ROLE_NAME]}) is True
assert is_viewer({"roles": [ADMIN_ROLE_NAME]}) is True
assert is_viewer({"roles": []}) is False

def test_group_names_roles_cover_writer_and_viewer_without_affinities_uuid(self):
with self._patch_uuid(""), self._patch_group_names(f"/ndp_ep/ep-{self.EP}"):
assert is_writer({"roles": [f"group:ndp_ep/ep-{self.EP}:writer"]}) is True
assert is_viewer({"roles": [f"group:ndp_ep/ep-{self.EP}:viewer"]}) is True

def test_effective_role_returns_highest_tier(self):
with self._patch_uuid(self.EP):
assert effective_role({"roles": []}) == "none"
Expand Down
17 changes: 17 additions & 0 deletions tests/test_get_current_user.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,17 @@
import importlib
from types import SimpleNamespace


def test_configured_test_token_bypasses_idp_with_ndp_admin(monkeypatch):
"""The local TEST_TOKEN must enter the normal admin authorization path."""
module = importlib.import_module("api.services.auth_services.get_current_user")
monkeypatch.setattr(
module, "swagger_settings", SimpleNamespace(test_token="local-token")
)

user = module.get_current_user(SimpleNamespace(credentials="local-token"))

assert user["sub"] == "test_user"
assert user["username"] == "Test User"
assert user["roles"] == ["ndp_admin"]
assert user["groups"] == []
13 changes: 8 additions & 5 deletions ui/src/services/api.js
Original file line number Diff line number Diff line change
Expand Up @@ -427,17 +427,20 @@ export const accessRequestsAPI = {

/**
* Return true if the given user_info payload grants admin access to the
* access-request management page. Admins are: holders of the `ndp_admin`
* realm role OR holders of any role ending in `_admin` (which covers the
* endpoint-scoped `{UUID}_admin` role).
* access-request management page. Prefer the backend's effective_role,
* and keep role-name checks as a fallback for older API responses.
*/
export const isAccessRequestAdmin = (userInfo) => {
if (userInfo?.effective_role === 'admin') return true;

const roles = userInfo?.roles;
if (!Array.isArray(roles)) return false;
return roles.some((role) => {
if (typeof role !== 'string') return false;
const lower = role.trim().toLowerCase();
return lower === 'ndp_admin' || lower.endsWith('_admin');
return lower === 'ndp_admin'
|| lower.endsWith('_admin')
|| (lower.startsWith('group:') && lower.endsWith(':admin'));
});
};

Expand Down Expand Up @@ -613,4 +616,4 @@ export const getApiBaseUrl = () => {
return BASE_URL;
};

export default apiClient;
export default apiClient;
25 changes: 25 additions & 0 deletions ui/src/services/api.test.js
Original file line number Diff line number Diff line change
@@ -0,0 +1,25 @@
import { isAccessRequestAdmin } from './api';

describe('isAccessRequestAdmin', () => {
it('uses effective_role when the backend provides it', () => {
expect(isAccessRequestAdmin({ roles: [], effective_role: 'admin' })).toBe(true);
});

it('accepts Keycloak group-path admin roles as a fallback', () => {
expect(
isAccessRequestAdmin({
roles: ['group:ndp_ep/ep-6a619d3f8b9242b94b015efb:admin'],
})
).toBe(true);
});

it('rejects endpoint group membership without an admin role', () => {
expect(
isAccessRequestAdmin({
roles: ['default-roles-ndp'],
groups: ['ndp_ep/ep-6a619d3f8b9242b94b015efb'],
effective_role: 'none',
})
).toBe(false);
});
});
Loading