rs_server_osam/tasks.md

<< Back to index

Main tasks executed by OSAM service.

apply_user_access_policy(user, current_rights)

Apply access policy over an user in ovh

Source code in docs/rs-server/services/osam/rs_server_osam/tasks.py

def apply_user_access_policy(user, current_rights):
    """
    Apply access policy over an user in ovh
    """
    msg = ""
    try:
        obs_user = get_keycloak_handler().get_obs_user_from_keycloak_username(user)
        if obs_user:
            get_ovh_handler().apply_user_access_policy(obs_user, current_rights)
            return True, {
                "detail": f"S3 access policy applied for the OVH account associated with the Keycloak user {user}",
            }
    except Exception as exc:  # pylint: disable = broad-exception-caught
        logger.error(f"Error while applying access policy for OVH user id {obs_user}. {exc}")
        msg = str(exc)
    return False, {
        "detail": "Failed to apply the access policy to the OVH account "
        f"associated with the Keycloak account {user}. {msg}",
    }

build_s3_rights(user_info)

Builds the S3 access rights structure for a user based on their Keycloak roles.

This function classifies roles into read, write, and download operations, then computes the corresponding access rights by matching them against a configmap.

Parameters:

Name	Type	Description	Default
user_info	dict	Dictionary containing user attributes, specifically the "keycloak_roles" key with a list of role strings.	required

Returns:

Name	Type	Description
dict		A dictionary with three keys: - "read": List of read-only access paths. - "read_download": List of read+download access paths. - "write_download": List of write+download access paths.

Source code in docs/rs-server/services/osam/rs_server_osam/tasks.py

@traced_function()
def build_s3_rights(user_info):  # pylint: disable=too-many-locals
    """
    Builds the S3 access rights structure for a user based on their Keycloak roles.

    This function classifies roles into read, write, and download operations, then computes
    the corresponding access rights by matching them against a configmap.

    Args:
        user_info (dict): Dictionary containing user attributes, specifically the "keycloak_roles" key
                          with a list of role strings.

    Returns:
        dict: A dictionary with three keys:
              - "read": List of read-only access paths.
              - "read_download": List of read+download access paths.
              - "write_download": List of write+download access paths.
    """
    # maybe we should use the user id instead of the username ?
    # Step 1: Parse roles
    read_roles = []
    write_roles = []
    download_roles = []

    for role in user_info["keycloak_roles"]:
        parsed = parse_role(role)
        if not parsed:
            continue
        owner, collection, op = parsed
        if op == "read":
            read_roles.append((owner, collection))
        elif op == "write":
            write_roles.append((owner, collection))
        elif op == "download":
            download_roles.append((owner, collection))

    # Step 2-3: Match against configmap
    read_set = match_roles(read_roles)
    write_set = match_roles(write_roles)
    download_set = match_roles(download_roles)

    # Step 3: Merge access
    read_only = read_set - download_set - write_set
    read_download = download_set
    write_download = write_set

    # Step 4: Output
    output = {
        STRKEY_ACCESS_RIGHT_READ_LIST: sorted(read_only),
        STRKEY_ACCESS_RIGHT_READ_DWN_LIST: sorted(read_download),
        STRKEY_ACCESS_RIGHT_WRITE_DWN_LIST: sorted(write_download),
    }

    logger.info(json.dumps(output, indent=2))
    return output

build_users_data_map()

Builds a dictionary mapping usernames to their associated user data.

For each user retrieved from Keycloak, this function gathers: - Custom attributes from Keycloak - Assigned Keycloak roles

Returns:

Name	Type	Description
dict	dict[str, Any]	A dictionary where each key is a username and the value is another dictionary containing: - "keycloak_attribute": Custom user attribute from Keycloak - "keycloak_roles": List of roles assigned to the user

Source code in docs/rs-server/services/osam/rs_server_osam/tasks.py

def build_users_data_map() -> dict[str, Any]:
    """
    Builds a dictionary mapping usernames to their associated user data.

    For each user retrieved from Keycloak, this function gathers:
      - Custom attributes from Keycloak
      - Assigned Keycloak roles

    Returns:
        dict: A dictionary where each key is a username and the value is another
              dictionary containing:
                - "keycloak_attribute": Custom user attribute from Keycloak
                - "keycloak_roles": List of roles assigned to the user
    """

    def build_user(kc_user: dict) -> tuple[str, dict]:
        """Process one user in a multithreaded context"""
        return (
            kc_user["username"],
            {
                "keycloak_attribute": get_keycloak_handler().get_obs_user_from_keycloak_user(kc_user),
                "keycloak_roles": [
                    role["name"] for role in get_keycloak_handler().get_keycloak_user_roles(kc_user["id"])
                ],
            },
        )

    kc_users = [(kc_user,) for kc_user in get_keycloak_handler().get_keycloak_users()]
    users_info = run_in_threads(build_user, kc_users, MAX_THREAD_COUNT, raise_exceptions=True)
    return dict(users_info)

create_obs_user_account_for_keycloak_user(keycloak_user)

Creates an OBS user and links it to a Keycloak user.

Parameters:

Name	Type	Description	Default
keycloak_user	dict	A dictionary representing the Keycloak user.	required

Returns:

Type	Description
	None

Source code in docs/rs-server/services/osam/rs_server_osam/tasks.py

@traced_function()
def create_obs_user_account_for_keycloak_user(
    keycloak_user: dict,
):
    """
    Creates an OBS user and links it to a Keycloak user.

    Args:
        keycloak_user (dict): A dictionary representing the Keycloak user.

    Returns:
        None
    """
    new_user_description = create_description_from_template(keycloak_user["username"], template=DESCRIPTION_TEMPLATE)
    new_user = get_ovh_handler().create_user(description=new_user_description, role=OVH_ROLE_FOR_NEW_USERS)
    get_keycloak_handler().set_obs_user_in_keycloak_user(keycloak_user, new_user["id"])

delete_obs_user_account_if_not_used_by_keycloak_account(obs_user, keycloak_users)

Deletes an OBS user if it is not linked to any Keycloak user.

Parameters:

Name	Type	Description	Default
obs_user	dict	Dictionary representing the OBS user.	required
keycloak_users	list[dict]	List of Keycloak user dictionaries.	required

Returns:

Type	Description
	None

Source code in docs/rs-server/services/osam/rs_server_osam/tasks.py

@traced_function()
def delete_obs_user_account_if_not_used_by_keycloak_account(
    obs_user: dict,
    keycloak_users: list[dict],
):
    """
    Deletes an OBS user if it is not linked to any Keycloak user.

    Args:
        obs_user (dict): Dictionary representing the OBS user.
        keycloak_users (list[dict]): List of Keycloak user dictionaries.

    Returns:
        None
    """
    if not all(val in obs_user["description"] for val in LIST_CHECK_OVH_DESCRIPTION):
        logger.info(f"The ovh user '{obs_user['username']}' is not created by osam service. Skipping....")
        return
    logger.info(f"Getting the keycloak username from ovh description '{obs_user['description']}'")
    logger.debug(f"DESCRIPTION_TEMPLATE = {DESCRIPTION_TEMPLATE}")
    keycloak_user_id = get_keycloak_user_from_description(obs_user["description"], template=DESCRIPTION_TEMPLATE)
    if not keycloak_user_id:
        logger.info(
            f"Failed to find keycloak username in the description ({obs_user['description']}) for ovh user "
            f"{obs_user['username']}. Skipping....",
        )
        return
    logger.debug(f"Keycloak username = {keycloak_user_id}")
    does_user_exist = False
    for keycloak_user in keycloak_users:
        if keycloak_user["username"] == keycloak_user_id:
            does_user_exist = True
            logger.debug(
                f"The keycloak user '{keycloak_user['username']}' does exist in keycloak, skipping the deletion in ovh",
            )
            break

    if not does_user_exist:
        # NOTE: this may seem strange considering that we retrieve the keycloak_user_id from
        # get_keycloak_user_from_description, but when the original description doesn't match
        # the template, get_keycloak_user_from_description returns the full description
        expected_description = create_description_from_template(keycloak_user_id, template=DESCRIPTION_TEMPLATE)
        logger.debug(f"Expected description: '{expected_description}'")
        if obs_user["description"] == expected_description:
            logger.info(
                f"Removal of the OVH user '{obs_user['username']}' with id {obs_user['id']} linked with a "
                f"removed keycloak user '{keycloak_user_id}'",
            )
            get_ovh_handler().delete_user(obs_user["id"])
        else:
            logger.info(
                f"The OVH user '{obs_user['username']}' with description '{obs_user['description']}' was not "
                f"created by using the current template '{DESCRIPTION_TEMPLATE}'. Skipping....",
            )

get_keycloak_handler()

Used to get a copy of Keycloak handler

Source code in docs/rs-server/services/osam/rs_server_osam/tasks.py

def get_keycloak_handler():
    """Used to get a copy of Keycloak handler"""
    return KeycloakHandler()

get_ovh_handler()

Used to get a copy of Ovh handler

Source code in docs/rs-server/services/osam/rs_server_osam/tasks.py

def get_ovh_handler():
    """Used to get a copy of Ovh handler"""
    return OVHApiHandler()

get_user_s3_credentials(user)

Retrieves the S3 access and secret keys for a given user.

Parameters:

Name	Type	Description	Default
user	str	The username for whom to retrieve S3 credentials.	required

Returns:

Name	Type	Description
dict	dict	A dictionary containing 'access_key', 'secret_key', 'endpoint', 'region'
	dict	for the user's S3 storage.

Source code in docs/rs-server/services/osam/rs_server_osam/tasks.py

def get_user_s3_credentials(user: str) -> dict:
    """
    Retrieves the S3 access and secret keys for a given user.

    Args:
        user (str): The username for whom to retrieve S3 credentials.

    Returns:
        dict: A dictionary containing 'access_key', 'secret_key', 'endpoint', 'region'
        for the user's S3 storage.
    """
    try:
        obs_user = get_keycloak_handler().get_obs_user_from_keycloak_username(user)

        if not obs_user:
            raise RuntimeError(f"No s3 credentials associated with {user}")

        if not (access_key := get_ovh_handler().get_user_s3_access_key(obs_user)):
            raise RuntimeError(f"Error reading user {user} from OVH.")

        secret_key = get_ovh_handler().get_user_s3_secret_key(obs_user, access_key)
        return {
            "access_key": access_key,
            "secret_key": secret_key,
            # NOTE: maybe get the endpoint url and region from another request ?
            # maybe: /cloud/project/{self.ovh_service_name}/storage/access")
            "endpoint": os.environ["S3_ENDPOINT"],
            "region": os.environ["S3_REGION"],
        }

    except Exception as exc:  # pylint: disable = broad-exception-caught
        raise RuntimeError(f"Error while getting s3 credentials for OVH user id {obs_user}") from exc

link_rspython_users_and_obs_users()

Coordinates linking between Keycloak users and OVH object storage (OBS) users.

• Retrieves Keycloak and OBS users.
• Optionally links or removes users based on whether mappings exist.

Note

The linking/unlinking logic is currently commented out and should be implemented based on specific integration rules.

Source code in docs/rs-server/services/osam/rs_server_osam/tasks.py

@traced_function()
def link_rspython_users_and_obs_users():
    """
    Coordinates linking between Keycloak users and OVH object storage (OBS) users.

    - Retrieves Keycloak and OBS users.
    - Optionally links or removes users based on whether mappings exist.

    Note:
        The linking/unlinking logic is currently commented out and should be implemented
        based on specific integration rules.
    """
    logger.info("Checking the link between keycloak users and ovh accounts. Creating ovh accounts if missing")

    # Get all OVH users and their IDs for quick lookup
    obs_users = get_ovh_handler().get_all_users()
    obs_user_ids = {str(obs_user["id"]) for obs_user in obs_users}

    def create_user(kc_user):
        """Create one user in a multithreaded context"""
        # For each Keycloak user, check if there's an associated OBS user
        obs_user_id = get_keycloak_handler().get_obs_user_from_keycloak_user(kc_user)
        obs_user_id = obs_user_id[0] if isinstance(obs_user_id, list) and obs_user_id else obs_user_id
        # If no associated OBS user or if the associated OBS user ID is not in OVH, create a new OBS user account
        if not obs_user_id or obs_user_id not in obs_user_ids:
            logger.info(f"Creating a new ovh account linked to keycloak user '{kc_user}'")
            create_obs_user_account_for_keycloak_user(kc_user)

    kc_users = [(kc_user,) for kc_user in get_keycloak_handler().get_keycloak_users()]
    run_in_threads(create_user, kc_users, MAX_THREAD_COUNT, raise_exceptions=True)

    # Refresh state after potential creations
    # After we create OBS account and update keycloak attributes, we need to refresh the keycloak_users and
    # obs_users lists to have the updated data for the deletion step
    kc_users = get_keycloak_handler().get_keycloak_users()
    obs_users = [(obs_user,) for obs_user in get_ovh_handler().get_all_users()]

    # If the cloud provider user is not linked with a keycloak account, remove it.
    def delete_user(obs_user: dict):
        """Delete one user in a multithreaded context"""
        delete_obs_user_account_if_not_used_by_keycloak_account(obs_user, kc_users)

    run_in_threads(delete_user, obs_users, MAX_THREAD_COUNT, raise_exceptions=True)

traced_function(name=None)

Decorator to trace the execution of a function using OpenTelemetry spans.

Parameters:

Name	Type	Description	Default
name	str	Custom name for the span. Defaults to the function's name.	None

Returns:

Name	Type	Description
Callable		A wrapped function with tracing enabled.

Source code in docs/rs-server/services/osam/rs_server_osam/tasks.py

def traced_function(name=None):
    """
    Decorator to trace the execution of a function using OpenTelemetry spans.

    Args:
        name (str, optional): Custom name for the span. Defaults to the function's name.

    Returns:
        Callable: A wrapped function with tracing enabled.
    """

    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            span_name = name or func.__name__
            with tracer.start_as_current_span(span_name) as span:
                span.set_attribute("function.name", func.__name__)
                return func(*args, **kwargs)

        return wrapper

    return decorator

update_s3_rights_lists(s3_rights)

Constructs the final user S3 access policy document for ovhbased on the provided access rights.

This function takes access permissions derived from a user's Keycloak roles and configmap and builds a structured S3 access policy document. The policy includes separate blocks for read, read+download, and write+download permissions, formatted according to OVH-compatible bucket and object prefixes.

The function generates individual policy statements for each type of permission

• It constructs bucket-level statements with prefix conditions (e.g., s3:prefix).
• It generates exact resource permissions using expanded ARN-based paths.
• It ensures that duplicated prefixes are not added redundantly.
• Invalid or malformed paths are ignored with a logged warning.

Parameters:

Name	Type	Description	Default
s3_rights	dict	A dictionary of categorized access paths per permission type. Expected keys include: - 'read': list of paths with read-only access. - 'read_download': list of paths with read + download access. - 'write_download': list of paths with write + download access.	required

Returns:

Name	Type	Description
dict		A complete S3 access policy document including version and statements, ready to be applied to an OVH S3 user.

Source code in docs/rs-server/services/osam/rs_server_osam/tasks.py

@traced_function()
def update_s3_rights_lists(s3_rights):  # pylint: disable=too-many-locals
    """
    Constructs the final user S3 access policy document for ovhbased on the provided access rights.

    This function takes access permissions derived from a user's Keycloak roles and configmap and builds
    a structured S3 access policy document. The policy includes separate blocks for read,
    read+download, and write+download permissions, formatted according to OVH-compatible
    bucket and object prefixes.

    The function generates individual policy statements for each type of permission:
      - It constructs bucket-level statements with prefix conditions (e.g., s3:prefix).
      - It generates exact resource permissions using expanded ARN-based paths.
      - It ensures that duplicated prefixes are not added redundantly.
      - Invalid or malformed paths are ignored with a logged warning.

    Args:
        s3_rights (dict): A dictionary of categorized access paths per permission type.
            Expected keys include:
                - 'read': list of paths with read-only access.
                - 'read_download': list of paths with read + download access.
                - 'write_download': list of paths with write + download access.

    Returns:
        dict: A complete S3 access policy document including version and statements,
              ready to be applied to an OVH S3 user.
    """

    # fields from the s3 access rights lists
    access_rights_list_keys = [
        (STRKEY_ACCESS_RIGHT_READ_LIST, BLOCK_LIST_READ_TEMPLATE),
        (STRKEY_ACCESS_RIGHT_READ_DWN_LIST, BLOCK_LIST_READ_DOWNLOAD_TEMPLATE),
        (STRKEY_ACCESS_RIGHT_WRITE_DWN_LIST, BLOCK_LIST_WRITE_DOWNLOAD_TEMPLATE),
    ]
    statements: list[dict[str, Any]] = []
    for key, block in access_rights_list_keys:  # pylint: disable=too-many-nested-blocks
        if not s3_rights.get(key):
            continue
        resources = []
        for access in s3_rights[key]:
            # get the bucket, owner and collection
            parts = access.strip().split("/")
            # protection against a wrong obs access policy
            if len(parts) < 3:
                logger.warning(f"Wrong obs policy access found: {access}")
                continue
            bucket = f"arn:aws:s3:::{parts[0]}"
            owner_collection = f"{parts[1]}/{parts[2]}/*"
            # ovh does not like */*/* format, so use */*
            if owner_collection == "*/*/*":
                owner_collection = "*/*"
            # check in the current statements
            found_in_template_bucket = False
            for stmt in statements:
                if bucket == stmt["Resource"]:
                    found_in_template_bucket = True
                    if owner_collection not in stmt["Condition"]["StringLike"]["s3:prefix"]:
                        stmt["Condition"]["StringLike"]["s3:prefix"].append(owner_collection)
                    break
            if not found_in_template_bucket:
                template_bucket: dict[str, Any] = copy.deepcopy(BLOCK_LIST_BUCKETS)
                template_bucket["Resource"] = bucket
                template_bucket["Condition"]["StringLike"]["s3:prefix"] = [owner_collection]
                statements.append(template_bucket)

            template: dict[str, Any] = copy.deepcopy(block)
            resource = f"{template['Resource'].replace('%placeholder%', access)}"
            # find the first "all" (*) and remove everything after it, because it's useless, and
            # moreover, ovh will not recognize the syntax
            # there should be at least one * char, the last one, see the template['Resource'], last char
            # so no need for protection in case the * char is not found
            resources.append(resource[: resource.find("*") + 1])

        template["Resource"] = resources
        statements.append(template)

    # Fill in main access policy template
    final_policy = copy.deepcopy(S3_ACCESS_RIGHTS_TEMPLATE)
    final_policy["Version"] = datetime.now(timezone.utc).strftime("%Y-%m-%d")
    final_policy["Statement"] = statements
    logger.info(json.dumps(final_policy, indent=2))
    return final_policy