# Copyright (c) 2022, 2025, Oracle and/or its affiliates. # # This program is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License, version 2.0, # as published by the Free Software Foundation. # # This program is designed to work with certain software (including # but not limited to OpenSSL) that is licensed under separate terms, as # designated in a particular file or component or in included license # documentation. The authors of MySQL hereby grant you an additional # permission to link the program and your derivative works with the # separately licensed software that they have either included with # the program or referenced in the documentation. # # This program is distributed in the hope that it will be useful, but # WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See # the GNU General Public License, version 2.0, for more details. # # You should have received a copy of the GNU General Public License # along with this program; if not, write to the Free Software Foundation, Inc., # 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA from mrs_plugin.lib import core, roles, schemas, content_sets, auth_apps, database import re import os from zipfile import ZipFile import pathlib def prompt_for_url_context_root(default=None): """Prompts the user for the url_context_root Returns: The url_context_root as str """ return core.prompt( "Please enter the context path for this service [/myService]: ", {'defaultValue': default if default else "/myService"}).strip() def prompt_for_service_protocol(default=None): """Prompts the user for the supported service protocols Returns: The service protocols as str """ protocols = core.prompt_for_list_item( item_list=[ "HTTP", "HTTPS", # "WEBSOCKET VIA HTTP", "WEBSOCKET VIA HTTPS" ], prompt_caption=( "Please select the protocol(s) the service should support " f"[{default if default else 'HTTP,HTTPS'}]: "), prompt_default_value=default if default else 'HTTP,HTTPS', print_list=True, allow_multi_select=True) return ','.join(protocols) def format_service_listing(services, print_header=False): """Formats the listing of MRS services Args: services (list): A list of services as dicts print_header (bool): If set to true, a header is printed Returns: The formatted list of services """ if print_header: output = (f"{'ID':>3} {'PATH':25} {'ENABLED':8} {'PROTOCOL(s)':20} " f"{'DEFAULT':9}\n") else: output = "" for i, item in enumerate(services, start=1): url = item.get('url_host_name') + item.get('url_context_root') output += (f"{i:>3} {url[:24]:25} " f"{'Yes' if item['enabled'] else '-':8} " f"{','.join(item['url_protocol'])[:19]:20} " f"{'Yes' if item['is_current'] else '-':5}") if i < len(services): output += "\n" return output def format_metadata(host_ctx, version): """Formats the service metadata details Args: host_ctx (str): The url root context path version (int): The version in the metadata audit log Returns: The metadata details in a tabular string """ return f"{'ID':>3} {'ROOT PATH':25} {'VERSION':15}\n{1:>3} {host_ctx[:24]:25} {version}" def add_service(session, url_host_name, service): if "options" in service: service["options"] = core.convert_json(service["options"]) else: service["options"] = { "headers": { "Access-Control-Allow-Credentials": "true", "Access-Control-Allow-Headers": "Content-Type, Authorization, X-Requested-With, Origin, X-Auth-Token", "Access-Control-Allow-Methods": "GET, POST, PUT, DELETE, OPTIONS" }, "http": { "allowedOrigin": "auto" }, "logging": { "exceptions": True, "request": { "body": True, "headers": True }, "response": { "body": True, "headers": True } }, "returnInternalErrorDetails": True } path = service.get("url_context_root").lower() if path == "/mrs": raise Exception( f'The REST service path `{path}` is reserved and cannot be used.') # If there is no id for the given host yet, create a host entry if service.get("url_host_id") is None: host = core.select(table="url_host", where=["name=?"] ).exec(session, [url_host_name if url_host_name else '']).first if host: service["url_host_id"] = host["id"] else: service["url_host_id"] = core.get_sequence_id(session) core.insert(table="url_host", values={ "id": service["url_host_id"], "name": url_host_name or '' } ).exec(session) service["id"] = core.get_sequence_id(session) # metadata column was only added in 3.0.0 current_version = core.get_mrs_schema_version(session) if current_version[0] <= 2: service.pop("metadata", None) if not core.insert(table="service", values=service).exec(session).success: raise Exception("Failed to add the new service.") return service["id"] def validate_service_path(session, path): """Ensures the given path is valid in any of the registered services. Args: session (object): The database session to use. path (str): The path to validate. Returns: service, schema, content_set as dict. """ if not path: return None, None, None service = None schema = None content_set = None # Match path against services and schemas all_services = get_services(session) for item in all_services: host_ctx = item.get("host_ctx") if host_ctx == path[: len(host_ctx)]: service = item if len(path) > len(host_ctx): sub_path = path[len(host_ctx) :] db_schemas = schemas.get_schemas( service_id=service.get("id"), session=session ) if db_schemas: for item in db_schemas: request_path = item.get("request_path") if request_path == sub_path[: len(request_path)]: schema = item break if not schema: content_sets_local = content_sets.get_content_sets( service_id=service.get("id"), session=session ) if content_sets_local: for item in content_sets_local: request_path = item.get("request_path") if request_path == sub_path[: len(request_path)]: content_set = item break if not schema and not content_set: raise ValueError(f"The given schema or content set was not found.") break if not service: raise ValueError(f"The given MRS service was not found.") return service, schema, content_set def delete_service(session, service_id): res = core.delete(table="service", where=["id=?"]).exec( session, params=[service_id]) if not res.success: raise Exception( f"The specified service with id {service_id} was not found.") def delete_services(session, service_ids): for service_id in service_ids: delete_service(session, service_id) def update_services(session, service_ids, value, merge_options=False): """Makes a given change to a MRS service Args: session: The database session to use service_ids: The list of service ids to change value: The value to be set as a dict for all values that will be changed merge_options: If set to True, specified options will be merged rather than overwritten Returns: The result message as string """ # Update all given services for service_id in service_ids: service = get_service(session, service_id=service_id) if service is None: raise Exception( f"The specified service with id {core.convert_id_to_string(service_id)} was not found.") if "url_host_name" in value: host = core.select(table="url_host", where="name=?" ).exec(session, [value["url_host_name"]]).first if host: host_id = host["id"] else: host_id = core.get_sequence_id(session) core.insert(table="url_host", values={ "id": host_id, "name": value["url_host_name"] } ).exec(session) del value["url_host_name"] value["url_host_id"] = host_id # metadata column was only added in 3.0.0 current_version = core.get_mrs_schema_version(session) if current_version[0] <= 2: value.pop("metadata", None) value.pop("published", None) # Reset an empty in_development.developers list to None in_development = value.get("in_development", None) if in_development is not None: developers = in_development.get("developers", None) if developers is not None and len(developers) == 0: value["in_development"] = None # Prepare the merge of options, if requested if merge_options: options = value.get("options", None) # Check if there are options set already, if so, merge the options if options is not None: row = core.MrsDbExec(""" SELECT options IS NULL AS options_is_null FROM `mysql_rest_service_metadata`.`service` WHERE id = ?""", [service_id]).exec(session).first if row and row["options_is_null"] == 1: merge_options = False else: value.pop("options") if value: core.update("service", sets=value, where=["id=?"]).exec(session, [service_id]) # Merge options if requested if merge_options and options is not None: core.MrsDbExec(""" UPDATE `mysql_rest_service_metadata`.`service` SET options = JSON_MERGE_PATCH(options, ?) WHERE id = ? """, [options, service_id]).exec(session) def query_services(session, service_id: bytes = None, url_context_root=None, url_host_name="", get_default=False, developer_list=None, auth_app_id=None): """Query MRS services Query the existing services. Filters may be applied as the 'service_id' or the 'url_context_root' with the 'url_host_name'. In the case no service is found, the default service may be fetched if the 'get_default' is set to True. To get the default service, don't set any other filters and set 'get_default' to True. Args: session (object): The database session to use. service_id: The id of the service url_context_root (str): The context root for this service get_default (bool): Whether to return the default service Returns: The list of found services. """ if url_context_root and not url_context_root.startswith('/'): raise Exception("The url_context_root has to start with '/'.") url_host_name = "" # no longer supported current_service_id = get_current_service_id(session) if not current_service_id: current_service_id = "0x00000000000000000000000000000000" wheres = [] params = [current_service_id] current_version = core.get_mrs_schema_version(session) if current_version[0] <= 2: # Build SQL based on which input has been provided sql = f""" SELECT se.id, se.enabled, se.url_protocol, h.name AS url_host_name, se.url_context_root, se.comments, se.options, se.url_host_id, CONCAT(h.name, se.url_context_root) AS host_ctx, CONCAT(h.name, se.url_context_root) AS full_service_path, se.auth_path, se.auth_completed_url, se.auth_completed_url_validation, se.auth_completed_page_content, se.id = ? as is_current, NULL AS in_development, NULL AS sorted_developers, se.name FROM `mysql_rest_service_metadata`.`service` se LEFT JOIN `mysql_rest_service_metadata`.url_host h ON se.url_host_id = h.id """ else: sql = f""" SELECT se.id, se.enabled, se.published, se.url_protocol, h.name AS url_host_name, se.url_context_root, se.comments, se.options, se.url_host_id, CONCAT(h.name, se.url_context_root) AS host_ctx, (SELECT CONCAT(COALESCE(CONCAT(GROUP_CONCAT(IF(item REGEXP '^[A-Za-z0-9_]+$', item, QUOTE(item)) ORDER BY item), '@'), ''), h.name, se.url_context_root) FROM JSON_TABLE( se.in_development->>'$.developers', '$[*]' COLUMNS (item text path '$') ) AS jt) AS full_service_path, se.auth_path, se.auth_completed_url, se.auth_completed_url_validation, se.auth_completed_page_content, se.metadata, se.parent_id, se.id = ? as is_current, se.in_development, (SELECT GROUP_CONCAT(IF(item REGEXP '^[A-Za-z0-9_]+$', item, QUOTE(item)) ORDER BY item) FROM JSON_TABLE( se.in_development->>'$.developers', '$[*]' COLUMNS (item text path '$') ) AS jt) AS sorted_developers, se.name, (SELECT JSON_ARRAYAGG(aa.name) FROM `mysql_rest_service_metadata`.`service_has_auth_app` sa2 JOIN `mysql_rest_service_metadata`.`auth_app` AS aa ON sa2.auth_app_id = aa.id WHERE sa2.service_id = se.id) AS auth_apps FROM `mysql_rest_service_metadata`.`service` se LEFT JOIN `mysql_rest_service_metadata`.url_host h ON se.url_host_id = h.id """ if auth_app_id is not None: sql += """ JOIN `mysql_rest_service_metadata`.`service_has_auth_app` sa ON se.id = sa.service_id AND sa.auth_app_id = ? """ params.append(auth_app_id) # Make sure that each user only sees the services that are either public or the user is a developer of # wheres.append("(in_development IS NULL OR " # "SUBSTRING_INDEX(CURRENT_USER(),'@',1) MEMBER OF(in_development->>'$.developers'))") if service_id: wheres.append("se.id = ?") params.append(service_id) elif url_context_root is not None and url_host_name is not None and developer_list is None: wheres.append("h.name = ?") wheres.append("url_context_root = ?") params.append(url_host_name) params.append(url_context_root) wheres.append("se.in_development IS NULL") elif get_default: # if nothing else is supplied and get_default is True, then get the default service wheres = ["se.id = ?"] params = [current_service_id, current_service_id] return core.MrsDbExec(sql + core._generate_where(wheres), params).exec(session).items having = "" if developer_list is not None: def quote(s): return f"'{s}'" # Build the sorted_developer string that matches the selected column, use same quoting as MySQL developer_list.sort() sorted_developers = ",".join( dev if re.match("^[A-Za-z0-9_-]*$", dev) else quote(re.sub(r"(['\\])", "\\\\\\1", dev, 0, re.MULTILINE)) for dev in developer_list) having = "\nHAVING h.name = ? AND url_context_root = ? AND sorted_developers = ?" params.append(url_host_name) params.append(url_context_root) params.append(sorted_developers) result = core.MrsDbExec( sql + core._generate_where(wheres) + having + "\nORDER BY se.url_context_root, h.name, sorted_developers", params).exec(session).items if len(result) == 0 and get_default: # No service was found s if we should get the default, then lets get it wheres = ["se.id = ?"] params = [current_service_id, current_service_id] result = core.MrsDbExec( sql + core._generate_where(wheres), params).exec(session).items return result def get_service(session, service_id: bytes = None, url_context_root=None, url_host_name=None, get_default=False, developer_list=None): """Gets a specific MRS service If no service is specified, the service that is set as current service is returned if it was defined before Args: session (object): The database session to use. service_id: The id of the service url_context_root (str): The context root for this service get_default (bool): Whether to return the default service Returns: The service as dict or None on error in interactive mode """ # url_host_name kept as a param for temporary backwards compat, but is no longer supported result = query_services(session, service_id=service_id, url_context_root=url_context_root, url_host_name="", get_default=get_default, developer_list=developer_list) return result[0] if len(result) == 1 else None def get_services(session): """Get a list of MRS services Args: session (object): The database session to use. Returns: List of dicts representing the services """ return query_services(session) def get_current_service(session): service_id = get_current_service_id(session) return get_service(session=session, service_id=service_id) def get_current_service_id(session): """Returns the current service Args: session (object): The database session to use. Returns: The current or default service or None if no default is set """ if not session: raise RuntimeError("A valid session is required.") config = core.ConfigFile() current_objects = config.settings.get("current_objects", []) # Try to find the settings for the connection which the service resides on connection_settings = list(filter(lambda item: item["connection"] == core.get_session_uri(session), current_objects)) if not connection_settings: return None return connection_settings[0].get("current_service_id") def set_current_service_id(session, service_id: bytes): if not session: raise RuntimeError("A valid session is required.") config = core.ConfigFile() current_objects = config.settings.get("current_objects", []) # Try to find the settings for the connection which the service resides on connection_settings = list(filter(lambda item: item["connection"] == core.get_session_uri(session), current_objects)) if connection_settings: # Found the settings for this host connection_settings[0]["current_service_id"] = service_id else: # The settings for this host do not exist yet....create them. current_objects.append({ "connection": core.get_session_uri(session), "current_service_id": service_id }) config.settings["current_objects"] = current_objects config.store() def get_service_create_statement(session, service: dict, include_database_endpoints: bool, include_static_endpoints: bool, include_dynamic_endpoints: bool) -> str: output = [] result = [] service_linked_auth_apps = [] service_linked_auth_apps = auth_apps.get_auth_apps(session, service["id"]) # create the service output.append(f'CREATE OR REPLACE REST SERVICE {service.get("host_ctx")}') if service.get("enabled") != 1: output.append(" DISABLED") if service.get("comments"): # ignore either None or empty output.append(f" COMMENT {core.squote_str(service.get("comments"))}") if service.get("published", False): output.append(f" PUBLISHED") auth = [] if service.get("auth_path") != "/authentication": auth.append(f' PATH {core.quote_auth_app(service.get("auth_path"))}') if service.get("auth_completed_url"): # ignore either None or empty auth.append(f' REDIRECTION {core.quote_str(service.get("auth_completed_url"))}') if service.get("auth_completed_url_validation"): # ignore either None or empty auth.append(f' VALIDATION {core.quote_str(service.get("auth_completed_url_validation"))}') if service.get("auth_completed_page_content"): # ignore either None or empty auth.append(f' PAGE CONTENT {core.quote_str(service.get("auth_completed_page_content"))}') if auth: # ignore either None or empty auth.insert(0, f" AUTHENTICATION") output.append("\n".join(auth)) if service.get("options"): output.append(core.format_json_entry("OPTIONS", service.get("options"))) if service.get("metadata"): output.append(core.format_json_entry("METADATA", service.get("metadata"))) for auth_app in service_linked_auth_apps: output.append(f" ADD AUTH APP {core.quote_auth_app(auth_app["name"])} IF EXISTS") result.append("\n".join(output) + ";") if include_database_endpoints: for role in roles.get_roles(session, service["id"], include_global=False): result.append(roles.get_role_create_statement(session, role)) result += [schemas.get_schema_create_statement(session, schema, True) for schema in schemas.get_schemas(session, service["id"]) if schema["schema_type"] != "SCRIPT_MODULE"] if include_static_endpoints or include_dynamic_endpoints: result += [content_sets.get_content_set_create_statement(session, content_set, include_dynamic_endpoints) for content_set in content_sets.get_content_sets(session, service["id"])] return "\n\n".join(result) def store_service_create_statement(session, service: dict, file_path: str, zip: bool, include_database_endpoints: bool=False, include_static_endpoints: bool = False, include_dynamic_endpoints: bool=False): file_content = get_service_create_statement(session, service, include_database_endpoints, include_static_endpoints, include_dynamic_endpoints) if zip and file_path.endswith(".zip"): file_path = file_path[:-len(".zip")] with open(file_path, "w") as f: f.write(file_content) if zip: with ZipFile(f"{file_path}.zip", "w") as f: f.write(file_path, arcname=pathlib.Path(file_path).name) os.remove(file_path) def get_service_sdk_data(session, service_id, binary_formatter=None): return database.get_sdk_service_data( session, service_id, binary_formatter=binary_formatter )