[sso-admin] Improve postup's idempotency

The class was only checking whether or not a specific token exists in
moodle, and it should ensure that it has access to the right permissions

Reported by:	@elena61
merge-requests/33/head
Evilham 2022-09-06 19:29:37 +02:00
parent 075529f472
commit 34761e028b
No known key found for this signature in database
GPG Key ID: AE3EE30D970886BF
1 changed files with 68 additions and 39 deletions

View File

@ -33,7 +33,7 @@ from datetime import datetime, timedelta
import psycopg2 import psycopg2
import yaml import yaml
from typing import TYPE_CHECKING from typing import TYPE_CHECKING, List, Set
if TYPE_CHECKING: if TYPE_CHECKING:
from admin.flaskapp import AdminFlaskApp from admin.flaskapp import AdminFlaskApp
@ -41,6 +41,30 @@ if TYPE_CHECKING:
from .postgres import Postgres from .postgres import Postgres
def moodle_functions() -> Set[str]:
return set(
[
"core_course_update_courses",
"core_user_get_users",
"core_user_get_users_by_field",
"core_user_update_picture",
"core_user_update_users",
"core_user_delete_users",
"core_user_create_users",
"core_cohort_get_cohort_members",
"core_cohort_add_cohort_members",
"core_cohort_delete_cohort_members",
"core_cohort_create_cohorts",
"core_cohort_delete_cohorts",
"core_cohort_search_cohorts",
"core_cohort_update_cohorts",
"core_role_assign_roles",
"core_role_unassign_roles",
"core_cohort_get_cohorts",
]
)
class Postup: class Postup:
def __init__(self, app: "AdminFlaskApp") -> None: def __init__(self, app: "AdminFlaskApp") -> None:
ready = False ready = False
@ -98,7 +122,7 @@ class Postup:
self.configure_tipnc() self.configure_tipnc()
self.add_moodle_ws_token(app) self.add_moodle_ws_token(app)
def select_and_configure_theme(self, theme : str="cbe") -> None: def select_and_configure_theme(self, theme: str = "cbe") -> None:
try: try:
self.pg.update( self.pg.update(
"""UPDATE "mdl_config" SET value = '%s' WHERE "name" = 'theme';""" """UPDATE "mdl_config" SET value = '%s' WHERE "name" = 'theme';"""
@ -158,47 +182,52 @@ class Postup:
exit(1) exit(1)
def add_moodle_ws_token(self, app: "AdminFlaskApp") -> None: def add_moodle_ws_token(self, app: "AdminFlaskApp") -> None:
srv_id = 3
try: try:
token = self.pg.select( mdl_service = self.pg.select(
"""SELECT * FROM "mdl_external_tokens" WHERE "externalserviceid" = 3""" """SELECT name FROM "mdl_external_services" WHERE name='dd admin';"""
)[0][1] )
app.config.setdefault("MOODLE_WS_TOKEN", token) if not mdl_service:
return self.pg.update(
except: """INSERT INTO "mdl_external_services" ("name", "enabled", "requiredcapability", "restrictedusers", "component", "timecreated", "timemodified", "shortname", "downloadfiles", "uploadfiles") VALUES
# log.error(traceback.format_exc()) ('dd admin', 1, '', 1, NULL, 1621719763, 1621719850, 'dd_admin', 0, 0);"""
pass )
try: active_functions_res = self.pg.select(
self.pg.update( 'SELECT functionname FROM "mdl_external_services_functions" '
"""INSERT INTO "mdl_external_services" ("name", "enabled", "requiredcapability", "restrictedusers", "component", "timecreated", "timemodified", "shortname", "downloadfiles", "uploadfiles") VALUES f'WHERE "externalserviceid" = {srv_id}'
('dd admin', 1, '', 1, NULL, 1621719763, 1621719850, 'dd_admin', 0, 0);""" )
active_functions: List[str] = [
a[0] if a and isinstance(a, tuple) else a for a in active_functions_res
]
missing_functions = moodle_functions().difference(active_functions)
if missing_functions:
missing_functions_values = ", ".join(
(f"({srv_id}, '{f}')" for f in missing_functions)
)
self.pg.update(
'INSERT INTO "mdl_external_services_functions" '
'("externalserviceid", "functionname") '
f"VALUES {missing_functions_values};"
)
mdl_service_user = self.pg.select(
f"SELECT externalserviceid FROM mdl_external_services_users WHERE externalserviceid={srv_id}"
)
if not mdl_service_user:
self.pg.update(
"""INSERT INTO "mdl_external_services_users" ("externalserviceid", "userid", "iprestriction", "validuntil", "timecreated") VALUES
(3, 2, NULL, NULL, 1621719871);"""
)
token_results = self.pg.select(
f"""SELECT token FROM "mdl_external_tokens" WHERE "externalserviceid" = {srv_id}"""
) )
self.pg.update( if token_results:
"""INSERT INTO "mdl_external_services_functions" ("externalserviceid", "functionname") VALUES token = token_results[0][0]
(3, 'core_course_update_courses'), app.config.setdefault("MOODLE_WS_TOKEN", token)
(3, 'core_user_get_users'), return
(3, 'core_user_get_users_by_field'),
(3, 'core_user_update_picture'),
(3, 'core_user_update_users'),
(3, 'core_user_delete_users'),
(3, 'core_user_create_users'),
(3, 'core_cohort_get_cohort_members'),
(3, 'core_cohort_add_cohort_members'),
(3, 'core_cohort_delete_cohort_members'),
(3, 'core_cohort_create_cohorts'),
(3, 'core_cohort_delete_cohorts'),
(3, 'core_cohort_search_cohorts'),
(3, 'core_cohort_update_cohorts'),
(3, 'core_role_assign_roles'),
(3, 'core_role_unassign_roles'),
(3, 'core_cohort_get_cohorts');"""
)
self.pg.update(
"""INSERT INTO "mdl_external_services_users" ("externalserviceid", "userid", "iprestriction", "validuntil", "timecreated") VALUES
(3, 2, NULL, NULL, 1621719871);"""
)
b32 = "".join( b32 = "".join(
random.choices( random.choices(