Skip to content

Commit 8311592

Browse files
MohammadIqbalAD-NHSMohammadIqbalAD-NHS
authored
[PRMT-72] Allow non-GP admittance to the NDR (#629)
1 parent 47a4cb9 commit 8311592

File tree

8 files changed

+360
-123
lines changed

8 files changed

+360
-123
lines changed

‎lambdas/enums/lambda_error.py‎

Lines changed: 8 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -151,13 +151,17 @@ def create_error_body(self, params: Optional[dict] = None) -> str:
151151
"err_code": "LIN_5008",
152152
"message": "SSM parameter values for PCSE user role may not exist",
153153
}
154-
LoginGpODS = {
154+
LoginGpOrgRoleCode = {
155155
"err_code": "LIN_5009",
156-
"message": "SSM parameter values for GP organisation role code may not exist",
156+
"message": "SSM parameter value for GP organisation role code may not exist",
157157
}
158-
LoginPcseODS = {
158+
LoginPcseOdsCode = {
159159
"err_code": "LIN_5010",
160-
"message": "SSM parameter values for PSCE ODS code may not exist",
160+
"message": "SSM parameter value for PCSE ODS code may not exist",
161+
}
162+
LoginItocOdsCodes = {
163+
"err_code": "LIN_5011",
164+
"message": "SSM parameter value for ITOC ODS code may not exist",
161165
}
162166

163167
"""

‎lambdas/services/login_service.py‎

Lines changed: 6 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -170,25 +170,19 @@ def generate_repository_role(self, organisation: dict, smartcard_role: str):
170170
in self.token_handler_ssm_service.get_smartcard_role_gp_admin()
171171
):
172172
logger.info("GP Admin: smartcard ODS identified")
173-
if self.has_role_org_role_code(
174-
organisation, self.token_handler_ssm_service.get_org_role_codes()[0]
175-
):
176-
return RepositoryRole.GP_ADMIN
173+
return RepositoryRole.GP_ADMIN
177174

178175
if (
179176
smartcard_role
180177
in self.token_handler_ssm_service.get_smartcard_role_gp_clinical()
181178
):
182179
logger.info("GP Clinical: smartcard ODS identified")
183-
if self.has_role_org_role_code(
184-
organisation, self.token_handler_ssm_service.get_org_role_codes()[0]
185-
):
186-
return RepositoryRole.GP_CLINICAL
180+
return RepositoryRole.GP_CLINICAL
187181

188182
if smartcard_role in self.token_handler_ssm_service.get_smartcard_role_pcse():
189183
logger.info("PCSE: smartcard ODS identified")
190-
if self.has_role_org_ods_code(
191-
organisation, self.token_handler_ssm_service.get_org_ods_codes()[0]
184+
if self.has_pcse_org_ods_code(
185+
organisation, self.token_handler_ssm_service.get_pcse_ods_code()
192186
):
193187
return RepositoryRole.PCSE
194188

@@ -201,16 +195,8 @@ def generate_repository_role(self, organisation: dict, smartcard_role: str):
201195
)
202196

203197
@staticmethod
204-
def has_role_org_role_code(organisation: dict, role_code: str) -> bool:
205-
if organisation["role_code"].upper() == role_code.upper():
206-
return True
207-
return False
208-
209-
@staticmethod
210-
def has_role_org_ods_code(organisation: dict, ods_code: str) -> bool:
211-
if organisation["org_ods_code"].upper() == ods_code.upper():
212-
return True
213-
return False
198+
def has_pcse_org_ods_code(organisation: dict, ods_code: str) -> bool:
199+
return organisation["org_ods_code"].upper() == ods_code.upper()
214200

215201
def issue_auth_token(
216202
self,

‎lambdas/services/ods_api_service.py‎

Lines changed: 61 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -1,18 +1,23 @@
1-
from typing import Dict, List, NamedTuple
1+
from typing import Any, Dict, List, NamedTuple, Optional
22

33
import requests
4+
from enums.lambda_error import LambdaError
45
from enums.repository_role import OrganisationRelationship
6+
from services.base.ssm_service import SSMService
57
from services.token_handler_ssm_service import TokenHandlerSSMService
68
from utils.audit_logging_setup import LoggingService
9+
from utils.constants.ssm import GP_ORG_ROLE_CODE
710
from utils.exceptions import (
811
OdsErrorException,
912
OrganisationNotFoundException,
1013
TooManyOrgsException,
1114
)
15+
from utils.lambda_exceptions import LoginException
1216

1317
logger = LoggingService(__name__)
1418

1519
token_handler_ssm_service = TokenHandlerSSMService()
20+
ssm_service = SSMService()
1621

1722

1823
class Organisation(NamedTuple):
@@ -51,32 +56,53 @@ def fetch_organisation_with_permitted_role(self, ods_code_list: list[str]) -> Di
5156
ods_code = ods_code_list[0]
5257
logger.info(f"ods_code selected: {ods_code}")
5358

59+
itoc_ods_codes = token_handler_ssm_service.get_itoc_ods_codes()
60+
61+
if ods_code in itoc_ods_codes:
62+
logger.info(f"ODS code {ods_code} is ITOC, returning org data")
63+
return parse_ods_response({}, "", "ITOC")
64+
5465
org_data = self.fetch_organisation_data(ods_code)
5566

5667
logger.info(f"Org Data: {org_data}")
5768

58-
pcse_ods = find_and_get_pcse_ods(ods_code)
69+
gp_org_role_code = get_user_gp_org_role_code(org_data)
70+
71+
if gp_org_role_code is not None:
72+
logger.info(f"ODS code {ods_code} is a GP, returning org data")
73+
icb_ods_code = find_icb_for_user(org_data["Organisation"])
74+
response = parse_ods_response(org_data, gp_org_role_code, icb_ods_code)
75+
return response
76+
77+
pcse_ods_code = token_handler_ssm_service.get_pcse_ods_code()
5978

60-
if pcse_ods is not None:
79+
if ods_code == pcse_ods_code:
6180
logger.info(f"ODS code {ods_code} is PCSE, returning org data")
6281
response = parse_ods_response(org_data, "", "PCSE")
6382
return response
6483

65-
gpp_org = find_and_get_gpp_org_code(org_data)
84+
allowed_ods_code_list = (
85+
token_handler_ssm_service.get_allowed_list_of_ods_codes()
86+
)
6687

67-
if gpp_org is not None:
68-
logger.info(f"ODS code {ods_code} is a GPP, returning org data")
88+
if ods_code in allowed_ods_code_list:
89+
logger.info(f"ODS code {ods_code} is in allowed list, returning org data")
6990
icb_ods_code = find_icb_for_user(org_data["Organisation"])
70-
response = parse_ods_response(org_data, gpp_org, icb_ods_code)
91+
primary_org_role_code = get_user_primary_org_role_code(org_data)
92+
response = parse_ods_response(org_data, primary_org_role_code, icb_ods_code)
7193
return response
7294

73-
logger.info(f"ODS code {ods_code} is not a GPP or PCSE, returning empty list")
95+
logger.info(
96+
f"ODS code {ods_code} is not a GP, PCSE, ITOC nor in allowed list, returning empty list"
97+
)
7498
return {}
7599

76100

77101
def parse_ods_response(org_data, role_code, icb_ods_code) -> dict:
78-
org_name = org_data["Organisation"]["Name"]
79-
org_ods_code = org_data["Organisation"]["OrgId"]["extension"]
102+
org_name = org_data.get("Organisation", {}).get("Name", "")
103+
org_ods_code = (
104+
org_data.get("Organisation", {}).get("OrgId", {}).get("extension", "")
105+
)
80106

81107
response_dictionary = {
82108
"name": org_name,
@@ -89,25 +115,36 @@ def parse_ods_response(org_data, role_code, icb_ods_code) -> dict:
89115
return response_dictionary
90116

91117

92-
def find_and_get_gpp_org_code(org_details):
93-
logger.info("Checking GPP Roles")
94-
json_roles: List[Dict] = org_details["Organisation"]["Roles"]["Role"]
118+
def get_user_gp_org_role_code(org_data: Dict[str, Any]) -> Optional[str]:
119+
logger.info("starting ssm request to retrieve GP organisation role code")
120+
gp_org_role_code = ssm_service.get_ssm_parameter(GP_ORG_ROLE_CODE)
95121

96-
org_role_codes = token_handler_ssm_service.get_org_role_codes()
97-
for json_role in json_roles:
98-
if json_role["id"] in org_role_codes:
99-
return json_role["id"]
100-
return None
122+
if gp_org_role_code:
123+
logger.info("Checking if GP organisation role is present")
124+
json_roles: List[Dict] = org_data["Organisation"]["Roles"]["Role"]
125+
for json_role in json_roles:
126+
if json_role["id"] == gp_org_role_code:
127+
return json_role["id"]
128+
return None
129+
130+
logger.error(
131+
LambdaError.LoginGpOrgRoleCode.to_str(),
132+
{"Result": "Unsuccessful login"},
133+
)
134+
raise LoginException(500, LambdaError.LoginGpOrgRoleCode)
101135

102136

103-
def find_and_get_pcse_ods(ods_code):
104-
logger.info("Checking PCSE Roles")
105-
if ods_code == token_handler_ssm_service.get_org_ods_codes()[0]:
106-
return ods_code
107-
return None
137+
def get_user_primary_org_role_code(org_data: Dict[str, Any]) -> str:
138+
logger.info("Checking if a primary organisation role is present")
139+
json_roles: List[Dict] = org_data["Organisation"]["Roles"]["Role"]
140+
141+
for json_role in json_roles:
142+
if "primaryRole" in json_role:
143+
return json_role["id"]
144+
return ""
108145

109146

110-
def find_icb_for_user(org_data):
147+
def find_icb_for_user(org_data: Dict[str, Any]) -> str:
111148
logger.info("Checking relationships")
112149
try:
113150
relationships: List[Dict] = org_data["Rels"]["Rel"]

‎lambdas/services/token_handler_ssm_service.py‎

Lines changed: 43 additions & 28 deletions
Original file line numberDiff line numberDiff line change
@@ -2,9 +2,11 @@
22
from services.base.ssm_service import SSMService
33
from utils.audit_logging_setup import LoggingService
44
from utils.constants.ssm import (
5+
ALLOWED_ODS_CODES_LIST,
56
GP_ADMIN_USER_ROLE_CODES,
67
GP_CLINICAL_USER_ROLE_CODE,
78
GP_ORG_ROLE_CODE,
9+
ITOC_ODS_CODES,
810
PCSE_ODS_CODE,
911
PCSE_USER_ROLE_CODE,
1012
)
@@ -93,40 +95,53 @@ def get_smartcard_role_pcse(self) -> list[str]:
9395
response = values.split(",")
9496
return response
9597

96-
def get_org_role_codes(self) -> list[str]:
97-
logger.info("starting ssm request to retrieve required org roles codes")
98-
params = self.get_ssm_parameters(
99-
[
100-
GP_ORG_ROLE_CODE,
101-
]
98+
def get_gp_org_role_code(self) -> str:
99+
logger.info("starting ssm request to retrieve GP organisation role code")
100+
response = self.get_ssm_parameter(GP_ORG_ROLE_CODE)
101+
102+
if response:
103+
return response
104+
105+
logger.error(
106+
LambdaError.LoginGpOrgRoleCode.to_str(),
107+
{"Result": "Unsuccessful login"},
102108
)
103-
response = [params.get(GP_ORG_ROLE_CODE)]
104-
if None in response:
105-
logger.error(
106-
LambdaError.LoginGpODS.to_str(),
107-
{"Result": "Unsuccessful login"},
108-
)
109-
raise LoginException(500, LambdaError.LoginGpODS)
110-
return response
109+
raise LoginException(500, LambdaError.LoginGpOrgRoleCode)
111110

112-
def get_org_ods_codes(self) -> list[str]:
113-
logger.info("starting ssm request to retrieve required org ods codes")
114-
params = self.get_ssm_parameters(
115-
[
116-
PCSE_ODS_CODE,
117-
]
111+
def get_pcse_ods_code(self) -> str:
112+
logger.info("starting ssm request to retrieve PCSE ODS code")
113+
response = self.get_ssm_parameter(PCSE_ODS_CODE)
114+
115+
if response:
116+
return response
117+
118+
logger.error(
119+
LambdaError.LoginPcseOdsCode.to_str(),
120+
{"Result": "Unsuccessful login"},
118121
)
119-
response = [params.get(PCSE_ODS_CODE)]
120-
if None in response:
121-
logger.error(
122-
LambdaError.LoginPcseODS.to_str(),
123-
{"Result": "Unsuccessful login"},
124-
)
125-
raise LoginException(500, LambdaError.LoginPcseODS)
126-
return response
122+
raise LoginException(500, LambdaError.LoginPcseOdsCode)
123+
124+
def get_itoc_ods_codes(self) -> str:
125+
logger.info("starting ssm request to retrieve ITOC ODS codes")
126+
response = self.get_ssm_parameter(ITOC_ODS_CODES)
127+
if response:
128+
return response
129+
130+
logger.error(
131+
LambdaError.LoginItocOdsCodes.to_str(),
132+
{"Result": "Unsuccessful login"},
133+
)
134+
raise LoginException(500, LambdaError.LoginItocOdsCodes)
127135

128136
def get_jwt_private_key(self) -> list[str]:
129137
logger.info("starting ssm request to retrieve NDR private key")
130138
return self.get_ssm_parameter(
131139
parameter_key="jwt_token_private_key", with_decryption=True
132140
)
141+
142+
def get_allowed_list_of_ods_codes(self) -> str:
143+
logger.info("starting ssm request to retrieve allowed list of ODS codes")
144+
response = self.get_ssm_parameter(ALLOWED_ODS_CODES_LIST)
145+
if not response:
146+
logger.warning("No ODS codes found in allowed list")
147+
return response

‎lambdas/tests/unit/services/test_login_service.py‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -248,7 +248,7 @@ def test_generate_repository_role_gp_admin(set_env, mocker):
248248
return_value=[user_role_code],
249249
)
250250
mocker.patch.object(
251-
TokenHandlerSSMService, "get_org_role_codes", return_value=[org_role_code]
251+
TokenHandlerSSMService, "get_gp_org_role_code", return_value=org_role_code
252252
)
253253

254254
login_service = LoginService()
@@ -276,7 +276,7 @@ def test_generate_repository_role_gp_clinical(set_env, mocker):
276276
return_value=[user_role_code],
277277
)
278278
mocker.patch.object(
279-
TokenHandlerSSMService, "get_org_role_codes", return_value=[org_role_code]
279+
TokenHandlerSSMService, "get_gp_org_role_code", return_value=org_role_code
280280
)
281281

282282
login_service = LoginService()
@@ -307,7 +307,7 @@ def test_generate_repository_role_pcse(set_env, mocker):
307307
TokenHandlerSSMService, "get_smartcard_role_pcse", return_value=user_role_code
308308
)
309309
mocker.patch.object(
310-
TokenHandlerSSMService, "get_org_ods_codes", return_value=[ods_code]
310+
TokenHandlerSSMService, "get_pcse_ods_code", return_value=ods_code
311311
)
312312

313313
login_service = LoginService()

0 commit comments

Comments
 (0)