Coverage for /usr/local/lib/python3.10/site-packages/tgclients/auth.py: 39%
94 statements
« prev ^ index » next coverage.py v7.4.4, created at 2024-04-02 16:49 +0000
« prev ^ index » next coverage.py v7.4.4, created at 2024-04-02 16:49 +0000
1# SPDX-FileCopyrightText: 2022 Georg-August-Universität Göttingen
2#
3# SPDX-License-Identifier: LGPL-3.0-or-later
5"""Provide access to the TextGrid Authorization Service."""
6import logging
7from typing import Optional, List
9from zeep import Client
10from zeep.exceptions import Fault, TransportError
12from tgclients.config import TextgridConfig
14TG_STANDARD_ROLE_MANAGER = 'Projektleiter'
15TG_STANDARD_ROLE_ADMINISTRATOR = 'Administrator'
16TG_STANDARD_ROLE_EDITOR = 'Bearbeiter'
17TG_STANDARD_ROLE_OBSERVER = 'Beobachter'
19logger = logging.getLogger(__name__)
20# prevent the zeep "Forcing soap:address location to HTTPS" warning
21logging.getLogger('zeep.wsdl.bindings.soap').setLevel(logging.ERROR)
24class TextgridAuth:
25 """Provide access to the TextGrid Authorization Service."""
27 def __init__(self, config: TextgridConfig = TextgridConfig()) -> None:
28 self._config = config
29 self._client = self._connect()
30 self._extra_crud_client = self._connect_extra_crud()
32 def _connect(self) -> Client:
33 """Internal helper that provides a SOAP client that is configured for
34 the use with the Textgrid Auth service.
36 Returns:
37 Client: A SOAP client
38 """
39 client = Client(self._config.auth_wsdl)
40 # this is a dirty hack; should be remediated
41 client.service._binding_options['address'] = self._config.auth_address
42 return client
44 def _connect_extra_crud(self) -> Client:
45 """Internal helper that provides a SOAP client that is configured for
46 the use with the Textgrid Auth service (the extra crud service).
48 Returns:
49 Client: A SOAP client
50 """
51 client = Client(self._config.extra_crud_wsdl)
52 # this is a dirty hack; should be remediated
53 client.service._binding_options['address'] = self._config.extra_crud_address
54 return client
56 def list_assigned_projects(self, sid: str) -> List[str]:
57 """Get assigned projects.
59 Args:
60 sid (str): Session ID
62 Raises:
63 TextgridAuthException: in case of transport exceptions
65 Returns:
66 List: A list of project id strings
67 """
68 try:
69 return self._client.service.tgAssignedProjects(sid)
70 except TransportError as error:
71 message = 'Error listing assigned projects. Is your sessionID valid?'
72 logger.warning(message)
73 raise TextgridAuthException(message) from error
75 def list_all_projects(self) -> List[str]:
76 """Returns all projects stored in this RBAC instance with ID, name,
77 and description. See also getProjectDescription(). SID
78 is not needed as this information can be reviewed publicly.
80 Returns:
81 list: list of each project with ID, name and description
82 """
83 # sid is optional, so we pass empty sid
84 return self._client.service.getAllProjects('')
86 def get_project_description(self, project_id: str):
87 """Returns name and description of project identified by ID. See also getAllProjects().
89 Args:
90 project_id (str): the project ID
92 Returns:
93 zeep.objects.projectInfo: project info with id, name and description
94 """
95 return self._client.service.getProjectDescription('', '', project_id)
97 def create_project(self, sid: str, name: str, description: str,
98 default_owner_roles: Optional[bool] = True) -> str:
99 """Create a new project
101 Args:
102 sid (str): TextGrid Session ID
103 name (str): name of the project
104 description (str): description for the project
105 default_owner_roles (bool, optional): whether to assign the default roles to the owner
106 (editor, authority to delete). Defaults to True.
108 Raises:
109 TextgridAuthException: in case of tgauth faults
111 Returns:
112 str: the project ID of the created project
113 """
114 try:
115 project_id = self._client.service.createProject(auth=sid, name=name,
116 description=description)
117 except Fault as fault:
118 message = 'Error creating project. Is your sessionID valid?'
119 logger.warning(message)
120 raise TextgridAuthException(message) from fault
122 if default_owner_roles:
123 eppn = self.get_eppn_for_sid(sid)
124 self.add_editor_to_project(sid, project_id, eppn)
125 self.add_admin_to_project(sid, project_id, eppn)
127 return project_id
129 def get_eppn_for_sid(self, sid: str) -> str:
130 """Get the EPPN belonging to a sessionID
132 Args:
133 sid (str): TextGrid Session ID
135 Raises:
136 TextgridAuthException: in case of transport exceptions
138 Returns:
139 str: the EPPN
140 """
141 try:
142 eppn = self._extra_crud_client.service.getEPPN(auth=sid, secret='')
143 except TransportError as error:
144 message = 'Error getting eppn. Is your sessionID valid?'
145 logger.warning(message)
146 raise TextgridAuthException(message) from error
147 return eppn
149 def delete_project(self, sid: str, project_id: str) -> bool:
150 """Delete a project
152 Args:
153 sid (str): TextGrid Session ID
154 project_id (str): the project ID
156 Raises:
157 TextgridAuthException: in case of tgauth faults
159 Returns:
160 bool: true in case of success
161 """
162 try:
163 status = self._client.service.deleteProject(
164 auth=sid, project=project_id)
165 except Fault as fault:
166 message = 'Error deleting project. Is your sessionID valid?'
167 logger.warning(message)
168 raise TextgridAuthException(message) from fault
169 return status
171 def add_admin_to_project(self, sid: str, project_id: str, eppn: str) -> bool:
172 """Give an user the admin role in a project
174 Args:
175 sid (str): TextGrid Session ID
176 project_id (str): the project ID
177 eppn (str): the eppn identifying the user
179 Raises:
180 TextgridAuthException: in case of tgauth faults
182 Returns:
183 bool: true in case of success
184 """
185 try:
186 return self._add_role_to_project(sid, project_id, eppn, TG_STANDARD_ROLE_ADMINISTRATOR)
187 except TextgridAuthException as exception:
188 raise exception
190 def add_editor_to_project(self, sid: str, project_id: str, eppn: str) -> bool:
191 """Give an user the editor role in a project
193 Args:
194 sid (str): TextGrid Session ID
195 project_id (str): the project ID
196 eppn (str): the eppn identifying the user
198 Raises:
199 TextgridAuthException: in case of tgauth faults
201 Returns:
202 bool: true in case of success
203 """
204 try:
205 return self._add_role_to_project(sid, project_id, eppn, TG_STANDARD_ROLE_EDITOR)
206 except TextgridAuthException as exception:
207 raise exception
209 def add_manager_to_project(self, sid: str, project_id: str, eppn: str) -> bool:
210 """Give an user the manager role in a project
212 Args:
213 sid (str): TextGrid Session ID
214 project_id (str): the project ID
215 eppn (str): the eppn identifying the user
217 Raises:
218 TextgridAuthException: in case of tgauth faults
220 Returns:
221 bool: true in case of success
222 """
223 try:
224 return self._add_role_to_project(sid, project_id, eppn, TG_STANDARD_ROLE_MANAGER)
225 except TextgridAuthException as exception:
226 raise exception
228 def add_observer_to_project(self, sid: str, project_id: str, eppn: str) -> bool:
229 """Give an user the observer role in a project
231 Args:
232 sid (str): TextGrid Session ID
233 project_id (str): the project ID
234 eppn (str): the eppn identifying the user
236 Raises:
237 TextgridAuthException: in case of tgauth faults
239 Returns:
240 bool: true in case of success
241 """
242 try:
243 return self._add_role_to_project(sid, project_id, eppn, TG_STANDARD_ROLE_OBSERVER)
244 except TextgridAuthException as exception:
245 raise exception
247 def _add_role_to_project(self, sid: str, project_id: str, eppn: str, role: str) -> bool:
248 rolename = role + ',' + project_id + ',Projekt-Teilnehmer'
249 try:
250 status = self._client.service.addMember(
251 auth=sid, username=eppn, role=rolename)
252 except Fault as fault:
253 message = 'Error adding role to project. Is your sessionID valid?'
254 logger.warning(message)
255 raise TextgridAuthException(message) from fault
256 return status
259class TextgridAuthException(Exception):
260 """Exception communicating with tgauth"""