magpie.adapter.magpieowssecurity

Attributes

ProviderSigninAPI

CONTENT_TYPE_JSON

LOGGER

Classes

LooseVersion

This class abstracts handling of a project's versions.

Service

Resource of service type.

Permission

Applicable Permission values (names) under certain Service and Resource.

MagpieOWSSecurity

Functions

evaluate_call(→ Any)

Evaluates the specified call with a wrapped HTTP exception handling. On failure, tries to call.

verify_param(→ None)

Evaluate various parameter combinations given the requested verification flags. Given a failing verification,

get_constant(→ magpie.typedefs.SettingValue)

Search in order for matched value of constant_name:

get_connected_session(→ sqlalchemy.orm.session.Session)

Retrieve the session attached to the request or recreated it to ensure it is open and within scoped transaction.

invalidate_service(→ None)

Invalidates any caching reference to the specified service name.

service_factory(→ ServiceInterface)

Retrieve the specific service class from the provided database service entry.

get_authenticate_headers(...)

Obtains all required headers by 401 responses based on executed request.

get_logger(→ logging.Logger)

Immediately sets the logger level to avoid duplicate log outputs from the root logger and this logger when

get_magpie_url(→ magpie.typedefs.Str)

Obtains the configured Magpie URL entrypoint based on the various combinations of supported configuration settings.

get_settings(→ magpie.typedefs.SettingsType)

Retrieve application settings from a supported container.

Module Contents

magpie.adapter.magpieowssecurity.evaluate_call(call: Callable[[], Any], fallback: Callable[[], None] | None = None, http_error: Type[pyramid.httpexceptions.HTTPError] = HTTPInternalServerError, http_kwargs: magpie.typedefs.ParamsType | None = None, msg_on_fail: magpie.typedefs.Str = '', content: magpie.typedefs.JSON | None = None, content_type: magpie.typedefs.Str = CONTENT_TYPE_JSON, metadata: magpie.typedefs.JSON | None = None) Any[source]

Evaluates the specified call with a wrapped HTTP exception handling. On failure, tries to call.

fallback if specified, and finally raises the specified http_error.

Any potential error generated by fallback or http_error themselves are treated as HTTPInternalServerError.

Exceptions are generated using the standard output method formatted based on specified content_type.

Example:

normal call:

try:
    res = func(args)
except Exception as exc:
    fb_func()
    raise HTTPExcept(exc.message)

wrapped call:

res = evaluate_call(lambda: func(args), fallback=lambda: fb_func(), http_error=HTTPExcept, **kwargs)
Parameters:
  • call – function to call, MUST be specified as lambda: <function_call>

  • fallback – function to call (if any) when call failed, MUST be lambda: <function_call>

  • http_error – alternative exception to raise on call failure

  • http_kwargs – additional keyword arguments to pass to http_error if called in case of HTTP exception

  • msg_on_fail – message details to return in HTTP exception if call failed

  • content – json formatted additional content to provide in case of exception

  • content_type – format in which to return the exception (one of magpie.common.SUPPORTED_ACCEPT_TYPES)

  • metadata – request metadata to add to the response body. (see: magpie.api.requests.get_request_info())

Raises:
  • http_error – on call failure

  • HTTPInternalServerError – on fallback failure

Returns:

whichever return value call might have if no exception occurred

magpie.adapter.magpieowssecurity.verify_param(param: Any, param_compare: Any | List[Any] | None = None, param_name: magpie.typedefs.Str | None = None, param_content: magpie.typedefs.JSON | None = None, with_param: bool = True, http_error: Type[pyramid.httpexceptions.HTTPError] = HTTPBadRequest, http_kwargs: magpie.typedefs.ParamsType | None = None, msg_on_fail: magpie.typedefs.Str = '', content: magpie.typedefs.JSON | None = None, content_type: magpie.typedefs.Str = CONTENT_TYPE_JSON, metadata: magpie.typedefs.JSON | None = None, not_none: bool = False, not_empty: bool = False, not_in: bool = False, not_equal: bool = False, is_true: bool = False, is_false: bool = False, is_none: bool = False, is_empty: bool = False, is_in: bool = False, is_equal: bool = False, is_type: bool = False, matches: bool = False) None[source]

Evaluate various parameter combinations given the requested verification flags. Given a failing verification, directly raises the specified http_error. Invalid usage exceptions generated by this verification process are treated as HTTPInternalServerError. Exceptions are generated using the standard output method.

Parameters:
  • param – parameter value to evaluate

  • param_compare – Other value(s) to test param against. Can be an iterable (single value resolved as iterable unless None). To test for None type, use is_none/not_none flags instead.

  • param_name – name of the tested parameter returned in response if specified for debugging purposes

  • param_content – Additional JSON content to apply to generated error content on raise when with_param is True. Must be JSON serializable. Provided content can override generated error parameter if matching fields.

  • with_param – On raise, adds values of param, param_name and param_compare, as well as additional failing conditions metadata to the JSON response body for each of the corresponding value.

  • http_error – derived exception to raise on test failure (default: HTTPBadRequest)

  • http_kwargs – additional keyword arguments to pass to http_error called in case of HTTP exception

  • msg_on_fail – message details to return in HTTP exception if flag condition failed

  • content – json formatted additional content to provide in case of exception

  • content_type – format in which to return the exception (one of magpie.common.SUPPORTED_ACCEPT_TYPES)

  • metadata – request metadata to add to the response body. (see: magpie.api.requests.get_request_info())

  • not_none – test that param is not None type

  • not_empty – test that param is not an empty iterable (string, list, set, etc.)

  • not_in – test that param does not exist in param_compare values

  • not_equal – test that param is not equal to param_compare value

  • is_true – test that param is True

  • is_false – test that param is False

  • is_none – test that param is None type

  • is_empty – test param for an empty iterable (string, list, set, etc.)

  • is_in – test that param exists in param_compare values

  • is_equal – test that param equals param_compare value

  • is_type – test that param is of same type as specified by param_compare type

  • matches – test that param matches the regex specified by param_compare value

Raises:
  • HTTPError – if tests fail, specified exception is raised (default: HTTPBadRequest)

  • HTTPInternalServerError – for evaluation error

Returns:

nothing if all tests passed

magpie.adapter.magpieowssecurity.ProviderSigninAPI[source]
class magpie.adapter.magpieowssecurity.LooseVersion(version: str)[source]

Bases: packaging.version.Version, VersionInterface

This class abstracts handling of a project’s versions.

A Version instance is comparison aware and can be compared and sorted using the standard Python interfaces.

>>> v1 = Version("1.0a5")
>>> v2 = Version("1.0")
>>> v1
<Version('1.0a5')>
>>> v2
<Version('1.0')>
>>> v1 < v2
True
>>> v1 == v2
False
>>> v1 > v2
False
>>> v1 >= v2
False
>>> v1 <= v2
True

Initialize a Version object.

Parameters:

version – The string representation of a version which will be parsed and normalized before use.

Raises:

InvalidVersion – If the version does not conform to PEP 440 in any way then this exception will be raised.

property version: Tuple[int | str, Ellipsis]
property patch
_cmp(other: LooseVersion | str) int[source]
magpie.adapter.magpieowssecurity.get_constant(constant_name: magpie.typedefs.Str, settings_container: magpie.typedefs.AnySettingsContainer | None = None, settings_name: magpie.typedefs.Str | None = None, default_value: magpie.typedefs.SettingValue | None = None, raise_not_set: bool = True, raise_missing: bool = True, print_missing: bool = False, empty_missing: bool = False) magpie.typedefs.SettingValue[source]
Search in order for matched value of constant_name:
  1. search in MAGPIE_CONSTANTS

  2. search in settings if specified

  3. search alternative setting names (see below)

  4. search in magpie.constants definitions

  5. search in environment variables

Parameter constant_name is expected to have the format MAGPIE_[VARIABLE_NAME] although any value can be passed to retrieve generic settings from all above-mentioned search locations.

If settings_name is provided as alternative name, it is used as is to search for results if constant_name was not found. Otherwise, magpie.[variable_name] is used for additional search when the format MAGPIE_[VARIABLE_NAME] was used for constant_name (i.e.: MAGPIE_ADMIN_USER will also search for magpie.admin_user and so on for corresponding constants).

Parameters:
  • constant_name – key to search for a value

  • settings_container – WSGI application settings container (if not provided, uses found one in current thread)

  • settings_name – alternative name for settings if specified

  • default_value – default value to be returned if not found anywhere, and exception raises are disabled.

  • raise_not_set – raise an exception if the found key is None, search until last case if others are None

  • raise_missing – raise exception if key is not found anywhere

  • print_missing – print message if key is not found anywhere, return None

  • empty_missing – consider an empty value for an existing key as if it was missing (i.e.: as if not set).

Returns:

found value or default_value

Raises:
  • ValueError – if resulting value is invalid based on options (by default raise missing/empty/None value)

  • LookupError – if no appropriate value could be found from all search locations (according to options)

magpie.adapter.magpieowssecurity.get_connected_session(request: pyramid.request.Request) sqlalchemy.orm.session.Session[source]

Retrieve the session attached to the request or recreated it to ensure it is open and within scoped transaction.

class magpie.adapter.magpieowssecurity.Service[source]

Bases: Resource

Resource of service type.

__tablename__ = 'services'
resource_id
resource_type_name = 'service'
__mapper_args__
property permissions
property url
property type
Identifier matching ``magpie.services.ServiceInterface.service_type``.
property sync_type
Identifier matching ``magpie.cli.SyncServiceInterface.sync_type``.
property configuration
Configuration modifiers for parsing access to resources and permissions.
static by_service_name(service_name, db_session)[source]
class magpie.adapter.magpieowssecurity.Permission[source]

Bases: magpie.utils.ExtendedEnum

Applicable Permission values (names) under certain Service and Resource.

READ = 'read'
WRITE = 'write'
ACCESS = 'access'
BROWSE = 'browse'
GET_CAPABILITIES = 'getcapabilities'
GET_MAP = 'getmap'
GET_FEATURE_INFO = 'getfeatureinfo'
GET_LEGEND_GRAPHIC = 'getlegendgraphic'
GET_METADATA = 'getmetadata'
GET_PROPERTY_VALUE = 'getpropertyvalue'
GET_FEATURE = 'getfeature'
GET_FEATURE_WITH_LOCK = 'getfeaturewithlock'
GET_GML_OBJECT = 'getgmlobject'
DESCRIBE_FEATURE_TYPE = 'describefeaturetype'
DESCRIBE_LAYER = 'describelayer'
DESCRIBE_PROCESS = 'describeprocess'
EXECUTE = 'execute'
LOCK_FEATURE = 'lockfeature'
TRANSACTION = 'transaction'
CREATE_STORED_QUERY = 'createstoredquery'
DROP_STORED_QUERY = 'dropstoredquery'
LIST_STORED_QUERIES = 'liststoredqueries'
DESCRIBE_STORED_QUERIES = 'describestoredqueries'
magpie.adapter.magpieowssecurity.invalidate_service(service_name: magpie.typedefs.Str) None[source]

Invalidates any caching reference to the specified service name.

magpie.adapter.magpieowssecurity.service_factory(service: magpie.models.Service, request: pyramid.request.Request) ServiceInterface[source]

Retrieve the specific service class from the provided database service entry.

magpie.adapter.magpieowssecurity.CONTENT_TYPE_JSON = 'application/json'[source]
magpie.adapter.magpieowssecurity.get_authenticate_headers(request: pyramid.request.Request, error_type: magpie.typedefs.Str = 'invalid_token') magpie.typedefs.HeadersType | None[source]

Obtains all required headers by 401 responses based on executed request.

Parameters:
  • request – request that was sent to attempt authentication or access which must respond with Unauthorized.

  • error_type – Additional detail of the cause of error. Must be one of (invalid_request, invalid_token, insufficient_scope).

magpie.adapter.magpieowssecurity.get_logger(name: magpie.typedefs.Str, level: int | None = None, force_stdout: bool = None, message_format: magpie.typedefs.Str | None = None, datetime_format: magpie.typedefs.Str | None = None) logging.Logger[source]

Immediately sets the logger level to avoid duplicate log outputs from the root logger and this logger when level is logging.NOTSET.

magpie.adapter.magpieowssecurity.get_magpie_url(container: magpie.typedefs.AnySettingsContainer | None = None) magpie.typedefs.Str[source]

Obtains the configured Magpie URL entrypoint based on the various combinations of supported configuration settings.

See also

Documentation section Application Settings for available setting combinations.

Parameters:

container – container that provides access to application settings.

Returns:

resolved Magpie URL

magpie.adapter.magpieowssecurity.get_settings(container: magpie.typedefs.AnySettingsContainer | None, app: bool = False) magpie.typedefs.SettingsType[source]

Retrieve application settings from a supported container.

Parameters:
  • container – supported container with a handle to application settings.

  • app – allow retrieving from current thread registry if no container was defined.

Returns:

found application settings dictionary.

Raises:

TypeError – when no application settings could be found or unsupported container.

magpie.adapter.magpieowssecurity.LOGGER[source]
class magpie.adapter.magpieowssecurity.MagpieOWSSecurity(container: magpie.typedefs.AnySettingsContainer)[source]

Bases: twitcher.interface.OWSSecurityInterface

_cached_request: Dict[uuid.UUID, pyramid.request.Request][source]
_get_service_cached(service_name: magpie.typedefs.Str, request_uuid: uuid.UUID) Tuple[magpie.services.ServiceInterface, Dict[str, magpie.typedefs.AnyValue]][source]

Cache this method with beaker based on the provided caching key parameters.

If the cache is not hit (expired timeout or new key entry), calls service_factory() to retrieve the actual ServiceInterface implementation. Otherwise, returns the cached service to avoid SQL queries.

Note

Function arguments are required to generate caching keys by which cached elements will be retrieved. Those arguments must be serializable to generate the cache key (i.e.: cannot pass a Request object that contains session and other unserializable/circular references).

get_service(request: pyramid.request.Request) magpie.services.ServiceInterface[source]

Obtains the service referenced by the request.

Caching is automatically handled according to configured application settings and whether the specific service name being requested was already processed recently and not expired.

verify_request(request: pyramid.request.Request) bool[source]

Verify that the service request is allowed.

Added in version 3.18: Available only in Twitcher >= 0.6.x.

check_request(request: pyramid.request.Request) None[source]

Verifies if the request user has access to the targeted resource according to parent service and permissions.

If the request path corresponds to configured Twitcher proxy, evaluate the ACL. Otherwise, ignore request access validation.

In the case Twitcher proxy path is matched, the Logged User MUST be allowed access following Effective Permissions resolution via ACL. Otherwise, :exception:`OWSAccessForbidden` is raised.

Failing to parse the request or any underlying component that raises an exception will be left up to the parent caller to handle the exception. In most typical use case, this means Twitcher will raise a generic :exception:`OWSException` with NoApplicableCode, unless the exception was more specifically handled.

Raises:
  • OWSAccessForbidden – If the user does not have access to the targeted resource under the service.

  • HTTPBadRequest – If a request parsing error was detected when trying to resolve the permission based on the service/resource.

  • Exception – Any derived exception that was not explicitly handled is re-raised directly after logging the event.

Returns:

Nothing if user has access.

update_request_cookies(request: pyramid.request.Request) None[source]

Ensure login of the user and update the request cookies if Twitcher is in a special configuration.

Only update if MAGPIE_COOKIE_NAME is missing and is retrievable from access_token field within the Authorization header. Counter-validate the login procedure by calling Magpie’s /session which should indicate if there is a logged user.