Copyright (C) 2020-2022 HERE Global B.V. and its affiliate(s).

All rights reserved.

This software and other materials contain proprietary information

controlled by HERE and are protected by applicable copyright legislation.

Any use and utilization of this software and other materials and

disclosure to any third parties is conditional upon having a separate

agreement with HERE for the access, use, utilization or disclosure of this

software. In the absence of such agreement, the use of the software is not

allowed.

"""
This module contains an :class:AAAAuthorizationApi class to perform oauth API operations.

The HERE API reference documentation used in this module can be found here:
|iam_api_reference|

.. |iam_api_reference| raw:: html

IAM API Reference # noqa
"""

import urllib.parse
from typing import Optional

from here.platform.api.base_api import BaseApi
from here.platform.auth import Auth
from here.platform.config import ApplicationConfig, PlatformConfig
from here.platform.exceptions import PlatformException

[docs]
class AAAAuthorizationApi(BaseApi):
"""
This class provides access to HERE platform AAA Authorization APIs.
"""

def init(
self,
base_url: str,
auth: Optional[Auth],
platform_config: PlatformConfig,
application_config: ApplicationConfig,
proxies: Optional[dict] = None,
):
"""
Instantiate API with auth token.

:param base_url: base url
:param auth: instance of Auth
:param platform_config: a mandatory :class:PlatformConfig object to provide
configuration information for the API.
:param application_config: a mandatory :class:ApplicationConfig object to provide
configuration information for the API.
:param proxies: an optional proxy configuration. Defaults to the environment proxy
configuration.
"""
super(AAAAuthorizationApi, self).init(
platform_config=platform_config,
application_config=application_config,
auth=auth,
proxies=proxies,
)
self.base_url = base_url

[docs]
def create_project(self, body: dict) -> dict:
"""
Create the requested Project.

:param body: a dictionary with fields id,name,description to create project.
:return: a dict with hrn of the created project.
:raises PlatformException: If platform responds with an HTTP error.
"""
path = "/projects"
url = self.format_url(self.base_url, path)
resp = self.post(url, json=body)
if resp.status_code == 201:
resp_json: dict = resp.json()
return resp_json
else:
raise PlatformException(resp)

[docs]
def delete_project(self, project: str):
"""
Delete the requested Project.

:param project: HRN identifying the project.
:raises PlatformException: If platform responds with an HTTP error.
"""
hrn_quote = urllib.parse.quote(project, safe="")
path = f"/projects/{hrn_quote}"
url = self.format_url(self.base_url, path)
resp = self.delete(url)
if resp.status_code == 202:
return
else:
raise PlatformException(resp)

[docs]
def get_project(self, project: str) -> dict:
"""
Get the requested Project.

:param project: HRN identifying the project.
:return: a dict with name,description of the project.
:raises PlatformException: If platform responds with an HTTP error.
"""
hrn_quote = urllib.parse.quote(project, safe="")
path = f"/projects/{hrn_quote}"
url = self.format_url(self.base_url, path)
resp = self.get(url)
if resp.status_code == 200:
resp_json: dict = resp.json()
return resp_json
else:
raise PlatformException(resp)

[docs]
def add_project_member(self, project: str, member: str):
"""
Add the member to the requested Project

:param project: HRN identifying the project.
:param member: HRN identifying the project member. Either user, app or group.
:raises PlatformException: If platform responds with an HTTP error.
"""
hrn_quote = urllib.parse.quote(project, safe="")
member_id_quote = urllib.parse.quote(member, safe="")
path = f"/projects/{hrn_quote}/members/{member_id_quote}"
url = self.format_url(self.base_url, path)
resp = self.post(url)
if resp.status_code in [200, 201]:
return
else:
raise PlatformException(resp)

[docs]
def get_roles(
self,
page_token: Optional[str] = None,
count: Optional[int] = 100,
role_name: Optional[str] = None,
resource: Optional[str] = None,
) -> dict:
"""
Retrieve the list of roles.

:param page_token: The cursor for pagination. Present only if there is an
additional page of data to view.
:param count: Number of records to return. Default is 100 records. Maximum is 100 records.
:param role_name: The name of the role to be returned in the result set.
:param resource: The hrn of the resource which the roles returned in the
result set should be associated with.
:return: a dict with roles information.
:raises PlatformException: If platform responds with an HTTP error.
"""
path = "/roles"
url = self.format_url(self.base_url, path)
params = {"pageToken": page_token, "count": count, "roleName": role_name, "resource": resource,}
resp = self.get(url, params=params)
if resp.status_code == 200:
resp_json: dict = resp.json()
return resp_json
else:
raise PlatformException(resp)

[docs]
def add_role_entity(self, role: str, entity: str):
"""
Assign the role provided to the given entity.

:param role: HRN identifying a given role
:param entity: HRN identifying a given entity
:raises PlatformException: If platform responds with an HTTP error.
"""
role_hrn_quote = urllib.parse.quote(role, safe="")
entity_quote = urllib.parse.quote(entity, safe="")
path = f"/roles/{role_hrn_quote}/entities/{entity_quote}"
url = self.format_url(self.base_url, path)
resp = self.post(url)
if resp.status_code == 204:
return
else:
raise PlatformException(resp)

[docs]
def patch_project(self, project: str, body: dict) -> dict:
"""
Update the specified Project.

:param project: HRN identifying the project.
:param body: a dictionary with fields name,description to update the project.
:return: a dict with project information.
:raises PlatformException: If platform responds with an HTTP error.
"""
hrn_quote = urllib.parse.quote(project, safe="")
path = f"/projects/{hrn_quote}"
url = self.format_url(self.base_url, path)
resp = self.patch(url, json=body)
if resp.status_code == 200:
resp_json: dict = resp.json()
return resp_json
else:
raise PlatformException(resp)

[docs]
def list_project_resources(
self,
project: str,
type: Optional[str] = None,
relation: Optional[str] = None,
limit: Optional[str] = None,
page_token: Optional[str] = None,
) -> dict:
"""
Get the list of resources in the requested Project.

:param project: HRN identifying the project.
:param type: The type of the resource.
:param relation: The relation of the resource. A resource is only
returned in the resp if it matches the requested relation.
:param limit: Number of entries to be returned in the resp.
:param page_token: The cursor for pagination. Present only if there
is an additional page of data to view.
:return: a dict with resources information.
:raises PlatformException: If platform responds with an HTTP error.
"""
hrn_quote = urllib.parse.quote(project, safe="")
path = f"/projects/{hrn_quote}/resources"
url = self.format_url(self.base_url, path)
params = {"type": type, "relation": relation, "limit": limit, "pageToken": page_token}
resp = self.get(url, params=params)
if resp.status_code == 200:
resp_json: dict = resp.json()
return resp_json
else:
raise PlatformException(resp)

[docs]
def delete_project_member(self, project: str, member: str):
"""
Remove the member from the specified Project.

:param project: HRN identifying the project.
:param member: HRN identifying the project member. Either user, app or group.
:raises PlatformException: If platform responds with an HTTP error.
"""
hrn_quote = urllib.parse.quote(project, safe="")
member_quote = urllib.parse.quote(member, safe="")
path = f"/projects/{hrn_quote}/members/{member_quote}"
url = self.format_url(self.base_url, path)
resp = self.delete(url)
if resp.status_code == 204:
return
else:
raise PlatformException(resp)

[docs]
def list_project_members(
self,
project: str,
only_include_identities: Optional[bool] = None,
limit: Optional[str] = None,
page_token: Optional[str] = None,
) -> dict:
"""
Get the list of members of the Project

:param project: HRN identifying the project.
:param only_include_identities: If true, returns an effective project
members list containing only user and app identities,
including those that are members of the project indirectly via a group.
It will also return users who are project admins of the specified project,
and Resource Managers for the realm.
Response will NOT include total number of identities. If false, returns users,
apps, and groups that are direct members of the project, excluding any users and apps
that only have membership via a group. Defaults to false.
:param limit: Number of entries to be returned in the resp.
:param page_token: The cursor for pagination. Present only if there is an
additional page of data to view.
:return: a dict with project members information.
:raises PlatformException: If platform responds with an HTTP error.
"""
hrn_quote = urllib.parse.quote(project, safe="")
path = f"/projects/{hrn_quote}/members"
url = self.format_url(self.base_url, path)
params = {"onlyIncludeIdentities": only_include_identities, "limit": limit, "pageToken": page_token,}
resp = self.get(url, params=params)
if resp.status_code == 200:
resp_json: dict = resp.json()
return resp_json
else:
raise PlatformException(resp)

[docs]
def get_project_list(
self,
limit: Optional[str] = None,
page_token: Optional[str] = None,
can_manage: Optional[bool] = None,
is_member: Optional[bool] = None,
) -> dict:
"""
Get the list of Projects you are a project admin or a member.

:param limit: Number of entries to be returned in the resp.
:param page_token: The cursor for pagination. Present only if
there is an additional page of data to view.
:param can_manage: if true returns all projects of which the
caller(user/app) is a project admin.
:param is_member: if true returns all projects of which the
caller(user/app) is a member.
:return: a dict with project information.
:raises PlatformException: If platform responds with an HTTP error.
:raises ValueError: If can_manage and is_member is not provided correctly.
"""
path = "/projects/me"
url = self.format_url(self.base_url, path)
if can_manage is True and is_member is True:
raise ValueError(
"Only one of 'canManage' / 'isMember' query parameter is supported at a time."
)
elif can_manage is True:
query = {"limit": limit, "canManage": str(can_manage).lower(), "pageToken": page_token}
elif is_member is True:
query = {"limit": limit, "isMember": str(is_member).lower(), "pageToken": page_token}
elif is_member is None and can_manage is None:
query = {"limit": limit, "pageToken": page_token}
else:
raise ValueError(
"can_manage and is_member not provided. "
"Set can_manage parameter to True to retrieve projects for which"
" the caller(user/app) is a project admin."
"Set is_member to True to to retrieve all projects of which"
" the caller(user/app) is a member"
)

resp = self.get(url, params=query)
if resp.status_code == 200:
resp_json: dict = resp.json()
return resp_json
else:
raise PlatformException(resp)

[docs]
def leave_project(self, project: str):
"""
Remove the caller from the specified Project.

:param project: HRN identifying the project.
:raises PlatformException: If platform responds with an HTTP error.
"""
project_hrn_quote = urllib.parse.quote(project, safe="")
path = f"/projects/{project_hrn_quote}/members/me"
url = self.format_url(self.base_url, path)
resp = self.delete(url)
if resp.status_code == 204:
return
else:
raise PlatformException(resp)

[docs]
def add_grant(
self, resource_hrn: str, entity_id: str, action_id: str, entity_type: str
) -> dict:
"""
Grant access to a resource to an entity

:param resource_hrn: The hrn that identifies the resource.
:param entity_id: The target entityId to grant access to.
:param action_id: The action to assign as allowed against
the resource as read,write,manage.
:param entity_type: The type of the entity to grant access to.
Must be one of user, app, or group
:return: a dict with grant related information.
:raises PlatformException: If platform responds with an HTTP error.
"""
path = f"/grants/resources/{resource_hrn}/entities/{entity_id}/actions/{action_id}"
url = self.format_url(self.base_url, path)
params = {"entityType": entity_type}
resp = self.post(url, params=params)
if resp.status_code == 200:
path = f"/grants/resources/{resource_hrn}:*/entities/{entity_id}/actions/{action_id}" # noqa: E501, E231
url = self.format_url(self.base_url, path)
resp = self.post(url, params=params)
if resp.status_code == 200:
resp_json: dict = resp.json()
return resp_json
else:
raise PlatformException(resp)

[docs]
def remove_grant(self, resource_hrn: str, entity_id: str, action_id: str, entity_type: str):
"""
Revoke access to a resource from an entity.

:param resource_hrn: The hrn that identifies the resource.
:param entity_id: The target entityId to revoke access from.
:param action_id: The action against the resource to
revoke access from as read,write,manage.
:param entity_type: The type of the entity to revoke access from.
Must be one of user, app, or group.
:raises PlatformException: If platform responds with an HTTP error.
"""
path = f"/grants/resources/{resource_hrn}/entities/{entity_id}/actions/{action_id}"
url = self.format_url(self.base_url, path)
params = {"entityType": entity_type}
resp = self.delete(url, params=params)
if resp.status_code == 204:
path = f"/grants/resources/{resource_hrn}:*/entities/{entity_id}/actions/{action_id}" # noqa: E501, E231
url = self.format_url(self.base_url, path)
resp = self.delete(url, params=params)
if resp.status_code == 204:
return
else:
raise PlatformException(resp)

As share permission isn't working with aaa authorization grant api, we invoke

the service directly here for now. This needs to be fixed once it's fixed in aaa

authorization api.

[docs]
def share_authorization(self, resource_hrn: str, entity_type: str, entity_id: str) -> dict:
"""
Share permission to a resource for an entity.

:param resource_hrn: The hrn that identifies the resource.
:param entity_id: The target entityId to revoke access from.
:param entity_type: The type of the entity to revoke access from.
Must be one of user, app, or group.
:return: Response dict with authorization info.
:raises PlatformException: If platform responds with an HTTP error.
"""
base_url = f"{self.platform_config.account_url}"
path = f"/{entity_type}/{entity_id}/authorization/share/permissions"
url = self.format_url(base_url, path)
data = {"permissions": [ { "action": "share", "effect": "allow", "resource": resource_hrn, "serviceId": "authorization", } ]}

resp = self.post(url=url, data=data)
if resp.status_code == 200:
data["permissions"][0]["resource"] = f"{resource_hrn}:*" # noqa: E231
resp = self.post(url=url, data=data)
if resp.status_code == 200:
resp_json: dict = resp.json()
return resp_json
else:
raise PlatformException(resp)