API reference¶

Reference of the app's Python code, generated from its docstrings. For the HTTP API, see API and the Swagger UI of a deployment.

App¶

app ¶

App factory, and entry point for development.

The image serves the app with Gunicorn (cloud_registry.app:create_app(), with FOCA's settings, foca.gunicorn_config); python -m cloud_registry.app runs it with Flask's development server.

create_app ¶

create_app() -> App

Create the app from the config, with the registry's checks.

Refuses service clients (FOCA's require_no_service_clients()), as the app doesn't support them, requires a size limit of request bodies and that the service info serves the registration rules of the app, adds missing keys of services, resumes interrupted renames of services (cf. resume_renames(); also before requests of the registry's operations, cf. resume_renames_periodically()) and adds the types of registered services that aren't stored (e.g., on upgrades). Warns of stored services with invalid URLs or URL schemes that are not allowed, and of rules that let owners change roles of groups although the rule for these is changed (cf. warn_owner_rules()).

FOCA validates the config, e.g., that token issuers are trusted (needed for access control) with an audience, the group settings and the service info (service_info), refuses to start without admins (security.access_control.require_admin), warns of placeholders in the service info, and limits the size of request bodies (server.max_content_length). Also requires the public URL of the app (server.public_url), for the Location header of new services.

Gunicorn calls this in each worker, after forking (MongoDB clients must not be shared across forks), so the checks run in each worker; if they fail, the worker fails to boot and Gunicorn exits.

Returns:

Type	Description
`App`	Connexion app.

Raises:

Type	Description
`RuntimeError`	Request bodies are not limited in size (`server.max_content_length` is null), the public URL (`server.public_url`) is not set, service clients are configured (which the app doesn't support), the service info doesn't serve the registration rules of the app, or stored services are invalid or duplicates.
`ValidationError`	The config is invalid, e.g., no token issuers are trusted, or the service info is invalid (naming the invalid field).
`ValueError`	The config is invalid, e.g., groups are mapped without a group claim, there are no admins (unless admins are optional, `security.access_control.require_admin`), the namespace of a trusted issuer starts with `groups/` (cf. `check_namespaces()`), or the custom config (`custom`) still sets the service info (`custom.endpoints.service_info`) or the size limit of request bodies (`custom.max_content_length`).

main ¶

main() -> None

Run the app with Flask's development server, on the configured port.

For development; not for production.

Configuration¶

custom_config ¶

Cloud Registry custom config models.

IdConfig ¶

Bases: FOCABaseConfig

Model for defining unique identifier for services on cloud registry.

Parameters:

Name	Type	Description	Default
`charset`		A string of allowed characters.	required
`length`		Length of returned string.	required

Attributes:

Name	Type	Description
`charset`	`str`	A string of allowed characters.
`length`	`PositiveInt`	Length of returned string.

Raises:

Type	Description
`ValidationError`	The class was instantianted with an illegal data type, or `charset` is empty, an expression (such as `string.digits`), which FOCA no longer evaluates, contains `/` or consists of dots only, as identifiers are URL path segments.

Example

IdConfig( ... charset='ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789', ... length=6 ... ) IdConfig(charset='ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789', length=6)

ServicesConfig ¶

Bases: FOCABaseConfig

Model for defining the service database store for cloud registry.

Defines the configurations for service identifiers stored on cloud registry, the schemes allowed in service URLs and the environments allowed for services.

Parameters:

Name	Description	Default
`id`	Unique identifier for a service in cloud registry.	required
`url_schemes`	Schemes allowed in service URLs (`url`), at least one of `http` and `https`; `https` only by default, as clients send access tokens to the registered services.	required
`environments`	Values allowed for the environment of services (`environment`), e.g., to keep filters consistent; by default, the values that the GA4GH specification suggests (`prod`, `test`, `dev`, `staging`). Empty to allow any value. At most 256 characters each, as in the API.	required

Attributes:

Name	Type	Description
`id`	`IdConfig`	Unique identifier for a service in cloud registry.
`url_schemes`	`list[Literal['http', 'https']]`	Schemes allowed in service URLs (`url`), at least one of `http` and `https`; `https` only by default, as clients send access tokens to the registered services.
`environments`	`list[Annotated[str, Field(max_length=256)]]`	Values allowed for the environment of services (`environment`), e.g., to keep filters consistent; by default, the values that the GA4GH specification suggests (`prod`, `test`, `dev`, `staging`). Empty to allow any value. At most 256 characters each, as in the API.

Raises:

Type	Description
`ValidationError`	The class was instantianted with an illegal data type, or `url_schemes` is empty or has other schemes than `http` and `https` (lowercase).

Example

ServicesConfig( ... id=IdConfig( ... charset='ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789', ... length=6 ... ) ... ) ServicesConfig(id=IdConfig(charset='ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789', length=6), url_schemes=['https'], environments=['prod', 'test', 'dev', 'staging'])

EndpointsConfig ¶

Bases: FOCABaseConfig

Model for defining endpoint configuration parameters.

Parameters:

Name	Type	Description	Default
`services`		Cloud registry service store parameters.	required

Attributes:

Name	Type	Description
`services`	`ServicesConfig`	Cloud registry service store parameters.

Raises:

Type	Description
`ValidationError`	The class was instantianted with an illegal data type.

Example

EndpointsConfig( ... services=ServicesConfig( ... id=IdConfig( ... charset='ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789', ... length=6 ... ) ... ) ... ) EndpointsConfig(services=ServicesConfig(id=IdConfig(charset='ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789', length=6), url_schemes=['https'], environments=['prod', 'test', 'dev', 'staging']))

AccessControlConfig ¶

Bases: FOCABaseConfig

Model for access control settings of the app.

Parameters:

Name	Type	Description	Default
`service_types`		Who may add service types: `open` (the default), any authenticated user, explicitly (`POST /services/types`) or by registering a service with a new type; `controlled`, admins only, and services of other users must have a registered type; `moderated`, any authenticated user, but types of users other than admins are pending until an admin approves them, and so are the services that have them.	required
`max_unused_service_types`		How many service types that no service has a user other than an admin may have added, in every mode (`409` above); unlimited if `None`. Bounds the lists of types, and of pending types, that users can grow.	required

Attributes:

Name	Type	Description
`service_types`	`Literal['open', 'controlled', 'moderated']`	Who may add service types: `open` (the default), any authenticated user, explicitly (`POST /services/types`) or by registering a service with a new type; `controlled`, admins only, and services of other users must have a registered type; `moderated`, any authenticated user, but types of users other than admins are pending until an admin approves them, and so are the services that have them.
`max_unused_service_types`	`NonNegativeInt \| None`	How many service types that no service has a user other than an admin may have added, in every mode (`409` above); unlimited if `None`. Bounds the lists of types, and of pending types, that users can grow.

Raises:

Type	Description
`ValidationError`	The class was instantianted with an illegal data type, `service_types` is not `open`, `controlled` or `moderated`, or `max_unused_service_types` is negative.

Example

AccessControlConfig(service_types='controlled') AccessControlConfig(service_types='controlled', max_unused_service_types=10)

CustomConfig ¶

Bases: FOCABaseConfig

Model for defining the custom configurations for cloud registry.

Parameters:

Name	Type	Description	Default
`endpoints`		Endpoint service configurations for cloud registry.	required
`access_control`		Access control settings of the app.	required

Attributes:

Name	Type	Description
`endpoints`	`EndpointsConfig`	Endpoint service configurations for cloud registry.
`access_control`	`AccessControlConfig`	Access control settings of the app.

Raises:

Type	Description
`ValidationError`	The class was instantianted with an illegal data type.

Example

CustomConfig( ... endpoints=EndpointsConfig( ... services=ServicesConfig( ... id=IdConfig( ... charset='ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789', ... length=6 ... ) ... ) ... ) ... ) CustomConfig(endpoints=EndpointsConfig(services=ServicesConfig(id=IdConfig(charset='ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789', length=6), url_schemes=['https'], environments=['prod', 'test', 'dev', 'staging'])), access_control=AccessControlConfig(service_types='open', max_unused_service_types=10))

utils ¶

Typed access to the configuration and collections of the app.

get_collection ¶

get_collection(name: str) -> Collection[dict[str, Any]]

Get a collection of the service store of the current app.

Parameters:

Name	Type	Description	Default
`name`	`str`	Name of the collection, as configured.	required

Returns:

Type	Description
`Collection[dict[str, Any]]`	MongoDB collection.

is_date_time ¶

is_date_time(value: object) -> bool

Check whether a string is an RFC 3339 date-time (non-strings pass).

Used for the timestamps of registered services (cf. cloud_registry.ga4gh.registry.service.check_date_times()).

now ¶

now() -> datetime

Get the current time, in the precision of the database.

Returns:

Type	Description
`datetime`	Current time (UTC), in milliseconds, like MongoDB dates.

format_time ¶

format_time(time: datetime) -> str

Format a time as an RFC 3339 date-time in UTC.

Used for the times of verifications and the timestamps of services.

Parameters:

Name	Type	Description	Default
`time`	`datetime`	Time in UTC; naive, as read from the database, or aware.	required

Returns:

Type	Description
`str`	RFC 3339 date-time in UTC, in milliseconds, e.g.,
`str`	`2026-10-20T12:34:56.789Z`.

Controllers¶

server ¶

Controllers for service endpoints.

getServices ¶

getServices(**kwargs: Any) -> list[dict[str, Any]]

List all services.

Parameters:

Name	Type	Description	Default
`**kwargs`	`Any`	Further arguments that Connexion passes (`user`, `token_info`); unused.	`{}`

Returns:

Type	Description
`list[dict[str, Any]]`	List of services, without the pending services that the user may
`list[dict[str, Any]]`	not see (cf. `visible_services()`).

getServiceById ¶

getServiceById(serviceId: str, **kwargs: Any) -> dict[str, Any]

Retrieve service by its identifier.

Parameters:

Name	Type	Description	Default
`serviceId`	`str`	Identifier of service to be retrieved.	required
`**kwargs`	`Any`	Further arguments that Connexion passes (`user`, `token_info`); unused.	`{}`

Returns:

Type	Description
`dict[str, Any]`	Service object.

Raises:

Type	Description
`NotFound`	The service doesn't exist, or is pending and the user may not see it (cf. `visible_services()`).

getServiceTypes ¶

getServiceTypes(**kwargs: Any) -> list[dict[str, Any]]

List types of services.

Parameters:

Name	Type	Description	Default
`**kwargs`	`Any`	Further arguments that Connexion passes (`user`, `token_info`); unused.	`{}`

Returns:

Type	Description
`list[dict[str, Any]]`	List of the stored service types (cf.
`list[dict[str, Any]]`	`cloud_registry.ga4gh.registry.service_types`), sorted by group,
`list[dict[str, Any]]`	artifact and version; without the pending types that the user may
`list[dict[str, Any]]`	not see (cf. `visible_types()`).

postService ¶

postService(**kwargs: Any) -> RegisterResponse

Add service with an auto-generated identifier.

Parameters:

Name	Type	Description	Default
`**kwargs`	`Any`	Further arguments that Connexion passes (`user`, `token_info`, and the request body); unused.	`{}`

Returns:

Type	Description
`RegisterResponse`	Identifier of registered service, with status `201` and its URL as
`RegisterResponse`	`Location`.

deleteService ¶

deleteService(serviceId: str, **kwargs: Any) -> str

Delete service.

The roles on the service are revoked first, so that none are left on a deleted service if deleting it fails, and nobody can act on it in between; and again afterwards, as roles granted concurrently (after their request checked that the service exists) would be left otherwise. Not while the service is renamed (cf. changing()).

Parameters:

Name	Type	Description	Default
`serviceId`	`str`	Identifier of service to be deleted.	required
`**kwargs`	`Any`	Further arguments that Connexion passes (`user`, `token_info`); unused.	`{}`

Returns:

Type	Description
`str`	Identifier of deleted service.

putService ¶

putService(serviceId: str, **kwargs: Any) -> RegisterResponse

Add/replace service with a user-supplied ID.

Only admins may add services: for others, the service must exist (cf. RegisterService._replace()), also if it is deleted concurrently. Not while the service is renamed (cf. changing()).

Parameters:

Name	Type	Description	Default
`serviceId`	`str`	Identifier of service to be registered/updated.	required
`**kwargs`	`Any`	Further arguments that Connexion passes (`user`, `token_info`, and the request body); unused.	`{}`

Returns:

Type	Description
`RegisterResponse`	Identifier of registered/updated service; for registered services,
`RegisterResponse`	with status `201` and their URL as `Location`.

service ¶

Controller for registering services.

HiddenDuplicateService ¶

Bases: CodedConflict

A service with the same URL and API type is registered (not named).

DuplicateService ¶

DuplicateService(existing_id: str)

Bases: CodedConflict

A service with the same URL and API type is registered.

The description names the registered service; for users who may not read it, responses use hidden() instead.

Attributes:

Name	Type	Description
`HIDDEN_DESCRIPTION`		Description without the registered service.
`existing_id`		Identifier of the registered service.

Create the exception for a service with the same URL and API type.

Parameters:

Name	Type	Description	Default
`existing_id`	`str`	Identifier of the registered service.	required

hidden ¶

hidden() -> HiddenDuplicateService

Get the conflict without the registered service.

Returns:

Type	Description
`HiddenDuplicateService`	Conflict whose description doesn't name the registered service,
`HiddenDuplicateService`	with the same code.

InvalidService ¶

Bases: BadRequest

The URL or API type of a service, or a link, is missing or invalid.

RegisterService ¶

RegisterService(data: dict[str, Any], id: str | None = None, allow_create: bool = True, keep_timestamps: bool = False)

Class for registering services with the registry.

Initialize service data.

The registry sets the timestamps of services (cf. DATE_TIME_FIELDS): createdAt when a service is registered, kept on updates, and updatedAt when it is registered or updated. Timestamps in data are ignored, unless keep_timestamps is set.

Parameters:

Name	Type	Description	Default
`data`	`dict[str, Any]`	Service metadata consistent with the `ExternalServiceRegister` schema.	required
`id`	`str \| None`	Service identifier. Auto-generated if not provided.	`None`
`allow_create`	`bool`	With an identifier, whether the service may be registered if it doesn't exist (e.g., for admins only, cf. `_replace()`); otherwise, only an existing service is replaced.	`True`
`keep_timestamps`	`bool`	Whether timestamps in `data` are stored as sent (e.g., for admins importing services), instead of those set by the registry; timestamps not in `data` are set as usual.	`False`

Attributes:

Name	Type	Description
`data`		Service metadata; a copy of `data`, with the identifier, and without timestamps unless `keep_timestamps` is set; the timestamps set by the registry are added on registration.
`replace`		Whether an existing service with the provided identifier should be replaced. Set to `True` if an `id` is provided, otherwise set to `False`.
`was_created`		Whether a new service was registered.
`verification_removed`		Whether the verification of a replaced service was removed (cf. `_replace()`).
`pending`		Whether the service is stored as pending (cf. `PENDING_SERVICE`); set before `register_metadata()`.
`allow_create`		Whether a service with the identifier may be registered.
`id_charset`		A string of allowed characters for generating service identifiers.
`id_length`		Length of generated service identifiers.
`url_schemes`		Schemes allowed in service URLs.
`environments`		Values allowed for `environment`; any if empty.
`db_coll`		Database collection for storing service objects.

validate ¶

validate() -> str

Check the service, before it is registered.

Also called by register_metadata(); callers may call it first, so that they only act (e.g., store its type) for valid services.

Returns:

Type	Description
`str`	Key of the service (cf. `service_key()`).

Raises:

Type	Description
`InvalidService`	Cf. `register_metadata()`.
`DuplicateService`	Another service with the same URL and API type is registered.

register_metadata ¶

register_metadata(retries: int = 9) -> None

Register service.

Parameters:

Name	Type	Description	Default
`retries`	`int`	Without a service identifier (POST), how many times the generation of a random identifier and insertion into the database are retried on `DuplicateKeyError`s. With an identifier (PUT), how many times the stored service is read again if it changed concurrently (cf. `_replace()`).	`9`

Raises:

Type	Description
`InvalidService`	The URL or API type of the service, the URL of a link, a timestamp (cf. `check_date_times()`) or a string (cf. `check_strings()`) is invalid, or the scheme of the service URL (cf. `check_url_scheme()`) or the environment (cf. `check_environment()`) is not allowed.
`DuplicateService`	Another service with the same URL and API type is registered.
`InternalServerError`	The service could not be registered within the retries.
`NotFound`	With an identifier, the service doesn't exist (also if it was deleted concurrently), and `allow_create` is false.

normalize_url ¶

normalize_url(url: str) -> str

Normalize a service URL for comparisons.

See foca.utils.urls.normalize_url(): common spellings are normalized, and queries and fragments removed; http and https URLs are different services.

Parameters:

Name	Type	Description	Default
`url`	`str`	Service URL; leading and trailing spaces are ignored.	required

Returns:

Type	Description
`str`	Normalized URL.

Raises:

Type	Description
`InvalidService`	The URL is invalid, or not an HTTP(S) URL.

check_url_scheme ¶

check_url_scheme(url: str, schemes: Collection[str]) -> None

Check that a service URL is valid and uses one of the allowed schemes.

See foca.utils.urls.check_url(), which checks the whole URL, like normalize_url(); for URLs that normalize_url() accepts, only the scheme can fail the check.

Parameters:

Name	Type	Description	Default
`url`	`str`	Service URL; leading and trailing spaces are ignored.	required
`schemes`	`Collection[str]`	Allowed schemes, lowercase (cf. `custom.endpoints.services.url_schemes`).	required

Raises:

Type	Description
`InvalidService`	The URL is invalid, or its scheme is not one of `schemes` (case-insensitive).

check_strings ¶

check_strings(value: object, field: str = '') -> None

Check that the strings of a request body can be stored.

Unpaired surrogates (e.g., from JSON escapes like \ud800) can't be encoded (UTF-8), so they fail when stored or logged.

Parameters:

Name	Type	Description	Default
`value`	`object`	Request body, or part of it.	required
`field`	`str`	Path of `value` in the body (e.g., `type.group`), for error messages.	`''`

Raises:

Type	Description
`InvalidService`	A string, or a key, has unpaired surrogates.

check_date_times ¶

check_date_times(data: dict[str, Any]) -> None

Check that the timestamps of a service are RFC 3339 date-times.

Parameters:

Name	Type	Description	Default
`data`	`dict[str, Any]`	Service metadata.	required

Raises:

Type	Description
`InvalidService`	A timestamp (cf. `DATE_TIME_FIELDS`) is not an RFC 3339 date-time.

check_environment ¶

check_environment(data: dict[str, Any], environments: list[str]) -> None

Check that the environment of a service is allowed.

Parameters:

Name	Type	Description	Default
`data`	`dict[str, Any]`	Service metadata.	required
`environments`	`list[str]`	Allowed values (cf. `custom.endpoints.services.environments`); any if empty.	required

Raises:

Type	Description
`InvalidService`	The service has an environment (`environment`) that is not one of `environments` (case-sensitive).

check_links ¶

check_links(data: dict[str, Any]) -> None

Check the URLs of the links of a service (cf. LINK_FIELDS).

Links are not compared, so not normalized, but checked as written (cf. foca.utils.urls.check_url()).

Parameters:

Name	Type	Description	Default
`data`	`dict[str, Any]`	Service metadata.	required

Raises:

Type	Description
`InvalidService`	A URL is invalid, or uses a scheme the field doesn't allow; the message names the field.

service_key ¶

service_key(data: dict[str, Any]) -> str

Get the key identifying a service by its URL and API type.

Parameters:

Name	Type	Description	Default
`data`	`dict[str, Any]`	Service metadata, with `url` and `type`.	required

Returns:

Type	Description
`str`	Key from the API type (`type.group`, `type.artifact`,
`str`	case-insensitive and without surrounding whitespace) and the
`str`	normalized URL, as a JSON array, but not the version, so that
`str`	upgrading a service to a new API version updates its entry.

Raises:

Type	Description
`InvalidService`	The URL or type is missing or invalid.

service_exists ¶

service_exists(id: str) -> bool

Check whether a service is registered.

Parameters:

Name	Type	Description	Default
`id`	`str`	Service identifier.	required

Returns:

Type	Description
`bool`	Whether a service with identifier `id` is registered.

add_service_keys ¶

add_service_keys() -> None

Add keys to services registered without one.

For example, to services registered before keys were introduced.

Concurrent runs of this function are safe (e.g., in each worker of the app on startup): keys added by another run are kept. Services registered without keys while it runs (e.g., by an app of an older version) are not safe: they may be missed or reported as duplicates.

Raises:

Type	Description
`RuntimeError`	Services have invalid URLs or types, or the same URL and API type; the message names all of them.

warn_invalid_urls ¶

warn_invalid_urls() -> None

Warn of stored services with keys whose URLs are invalid.

For example, URLs accepted before stricter checks were introduced (cf. "Stored services" in docs/upgrading.md). Such services are still served, but can only be updated with a valid URL, and their keys may differ from those of the same URL spelled validly, so registrations of the service with such a spelling are not detected as duplicates. Services without keys are checked when keys are added (cf. add_service_keys()).

warn_environments ¶

warn_environments() -> None

Warn of stored services whose environments are not allowed.

For example, of services registered before the environments were checked (custom.endpoints.services.environments). Such services are still served, but can only be updated with an allowed environment (or none).

warn_url_schemes ¶

warn_url_schemes() -> None

Warn of stored services whose URLs use schemes that are not allowed.

For example, http URLs registered before https became the only scheme allowed by default (custom.endpoints.services.url_schemes). Such services are still served, but can only be updated with an allowed scheme. Services with URLs that aren't strings or valid URLs (cf. normalize_url()) are skipped.

service_info ¶

Service info fields of the registry, computed from the config.

FOCA serves the service info (GET /service-info) from the config (service_info.service), with the fields that runtime_fields() returns (service_info.runtime_fields in config.yaml; called on each request): the rules for registrations (registration_rules), an extension of the GA4GH Service schema (RegistryServiceInfo in api/additions.openapi.yaml). They are built from the config in effect, so that clients (e.g., UIs) can read them instead of configuring them: the environments (custom.endpoints.services.environments) and service URL schemes (url_schemes) allowed, who may add service types (custom.access_control.service_types), and how many unused types users other than admins may have added (max_unused_service_types).

registration_rules ¶

registration_rules() -> dict[str, Any]

Get the rules for registrations of the config in effect.

Must be called in an app context.

Returns:

Type	Description
`dict[str, Any]`	Environments allowed for services (`environments`; empty if any
`dict[str, Any]`	value is), schemes allowed in service URLs (`url_schemes`), who may
`dict[str, Any]`	add service types (`service_types`: `open`, `controlled` or
`dict[str, Any]`	`moderated`), and how many types that no service has users other
`dict[str, Any]`	than admins may have added (`max_unused_service_types`; left out if
`dict[str, Any]`	there is no limit; `0` if they can't add types). Lists are copies.

runtime_fields ¶

runtime_fields() -> dict[str, Any]

Get the service info fields computed at runtime.

Called by FOCA in an app context, on startup and for each request for the service info (service_info.runtime_fields).

Returns:

Type	Description
`dict[str, Any]`	The registration rules (cf. `registration_rules()`).

check_service_info_fields ¶

check_service_info_fields() -> None

Check that the service info has the registration rules of the app.

Must be called in an app context.

Raises:

Type	Description
`RuntimeError`	No service info is configured, its fields aren't computed by `runtime_fields()` (`service_info.runtime_fields`), or the config sets the registration rules (`service_info.service.registration_rules`, also to `null`), which would take precedence over those of the app, so that they could differ from the rules the registry enforces.

verification ¶

Controllers for verifications of services.

Admins verify services they checked (e.g., that they accept tokens of the deployment's identity provider). The time of the verification is stored in an internal field of the service (VERIFIED), not served with the service, so that the GA4GH endpoints are unchanged. Replacing a service with another URL (as written) or API type removes it (cf. RegisterService._replace()), deleting a service deletes it with the service. Verifications are served with the URL and API type of the service, read from the same document, so that clients can match them against the services they list.

Only admins may verify services and remove verifications: no policy in config.yaml allows these operations to other users. Verifications can be read by anyone, except those of pending services that the user may not see (cf. cloud_registry.ga4gh.registry.service_types.visible_services()).

VerificationConflict ¶

VerificationConflict()

Bases: CodedConflict

The service has another URL or API type than the one checked.

Create the exception, with a description of what to do.

PendingVerification ¶

PendingVerification()

Bases: CodedConflict

The service is pending, so it can't be verified.

Create the exception, with a description of what to do.

getServiceVerification ¶

getServiceVerification(serviceId: str, **kwargs: Any) -> dict[str, Any]

Get the verification of a service.

Parameters:

Name	Type	Description	Default
`serviceId`	`str`	Service identifier.	required
`**kwargs`	`Any`	Further arguments that Connexion passes (`user`, `token_info`); unused.	`{}`

Returns:

Type	Description
`dict[str, Any]`	Verification of the service.

Raises:

Type	Description
`NotFound`	The service doesn't exist, is not verified, or is pending and the user may not see it.

putServiceVerification ¶

putServiceVerification(serviceId: str, verification: dict[str, Any] | None = None, **kwargs: Any) -> dict[str, Any] | Response

Verify a service, or renew its verification.

If the request names the URL and API type the admin checked, the service is only verified if it has them (checked in the same write), so that a service changed after the admin checked it isn't verified.

Parameters:

Name	Type	Description	Default
`serviceId`	`str`	Service identifier.	required
`verification`	`dict[str, Any] \| None`	URL and API type the admin checked; optional.	`None`
`**kwargs`	`Any`	Further arguments that Connexion passes (`user`, `token_info`); unused.	`{}`

Returns:

Type	Description
`dict[str, Any] \| Response`	Verification of the service, at the current time (or the later time
`dict[str, Any] \| Response`	of a previous verification); or a problem details response: `400`
`dict[str, Any] \| Response`	if the body is neither JSON nor empty, or the URL in it is invalid;
`dict[str, Any] \| Response`	`409` if the service has another URL or API type than the one
`dict[str, Any] \| Response`	checked, or is pending (cf. `PENDING_SERVICE`): services leave the
`dict[str, Any] \| Response`	pending state (their type is approved) before they are verified.

Raises:

Type	Description
`NotFound`	The service doesn't exist.

deleteServiceVerification ¶

deleteServiceVerification(serviceId: str, **kwargs: Any) -> str

Remove the verification of a service.

Parameters:

Name	Type	Description	Default
`serviceId`	`str`	Service identifier.	required
`**kwargs`	`Any`	Further arguments that Connexion passes (`user`, `token_info`); unused.	`{}`

Returns:

Type	Description
`str`	Identifier of the service.

Raises:

Type	Description
`NotFound`	The service doesn't exist or is not verified.

getVerifications ¶

getVerifications(**kwargs: Any) -> list[dict[str, Any]]

List verified services.

Parameters:

Name	Type	Description	Default
`**kwargs`	`Any`	Further arguments that Connexion passes (`user`, `token_info`); unused.	`{}`

Returns:

Type	Description
`list[dict[str, Any]]`	Identifier and verification of each verified service, sorted by
`list[dict[str, Any]]`	identifier; without the pending services that the user may not see.

Access control¶

permissions ¶

Access control: policies, service ownership and permission controllers.

The controllers manage permissions on services. Roles (security.access_control.resource_roles) are granted to users (by their identifier) or to groups (values of the group claim, security.access_control.claim_groups); groups count like users, e.g., a group may be the only owner of a service. Roles are granted, revoked and listed with FOCA's resource ownership helpers, which keep the last owner of a service (409 Conflict) and log changes of roles (who changed which role of whom on which service), for auditing. The app adds the display names of users, the check that services exist, and which groups users may grant roles to.

Policies are configured in config.yaml (security.access_control) and enforced by FOCA on all operations that require authentication, using the model in cloud_registry/access_control/model.conf. Subjects with the role admin (mapped from group claims via security.access_control.group_mapping, which deployments set) may do everything.

validate_bearer_token ¶

validate_bearer_token(token: str) -> dict[str, Any]

Validate a bearer token, keeping why it was rejected.

The function that Connexion validates tokens with (x-bearerInfoFunc in config.yaml): FOCA's validate_token(), but its exception is also kept for the request, so that check_read_token() can tell why a token of a public read was rejected without validating it again.

Parameters:

Name	Type	Description	Default
`token`	`str`	Bearer token.	required

Returns:

Type	Description
`dict[str, Any]`	Token info (cf. FOCA's `validate_token()`).

Raises:

Type	Description
`Unauthorized`	The token was rejected.

check_read_token ¶

check_read_token(func: Callable[..., T]) -> Callable[..., T]

Reject invalid tokens sent with reads that don't require login.

Public reads accept requests without a token (anonymous) and with one (for the visibility rules of the user; cf. security in the API specification). Connexion treats requests whose token it can't validate as anonymous then; so that clients notice (e.g., an expired token) and log in again, rather than see less without a hint, requests with an invalid token are rejected as on other operations (401, with an invalid_token challenge), with the exception kept when Connexion validated it (cf. validate_bearer_token(); validated again only if Connexion didn't, e.g., for an empty token). If it can't be validated for other reasons (e.g., the identity provider is unavailable), the request stays anonymous, so that public reads keep working, and this is logged.

Parameters:

Name	Type	Description	Default
`func`	`Callable[..., T]`	Controller of a read.	required

Returns:

Type	Description
`Callable[..., T]`	The controller, checking the token first.

services_path ¶

services_path() -> str

Get the path of the services, under which roles are granted.

Returns:

Type	Description
`str`	Request path of the service collection, including the base path of
`str`	the API handling the current request.

service_path ¶

service_path(serviceId: str) -> str

Get resource path of a service, i.e., the path roles are granted on.

Parameters:

Name	Type	Description	Default
`serviceId`	`str`	Service identifier.	required

Returns:

Type	Description
`str`	Request path of the service, including the base path of the API
`str`	handling the current request.

may_read_service ¶

may_read_service(serviceId: str) -> bool

Check whether the current user may get a service (GET).

The same decision as for a request to GET /services/{serviceId} with the token of the current request, and the opt-in roles it asks for (e.g., admin mode; security.access_control.opt_in_header): the route of that request decides whether FOCA enforces policies on it (only if reads require login, cf. "Requiring login for reads" in docs/access-control.md), and if so, its policies are evaluated by FOCA's is_permitted(), for the path and method of that request.

The check relies on internals of FOCA (_enforces_access_control()) and Connexion (connexion_context), as FOCA has no public helper for the decision on another request; the tests pin it to that of GET.

Parameters:

Name	Type	Description	Default
`serviceId`	`str`	Service identifier.	required

Returns:

Type	Description
`bool`	Whether the current user may get the service; `False` if the request
`bool`	isn't routed to `GET /services/{serviceId}` for the identifier (e.g.,
`bool`	for identifiers with slashes, which are routed to other operations or
`bool`	none).

may_add_service_types ¶

may_add_service_types() -> bool

Check whether the current user may add service types.

Must be called in a request context.

Returns:

Type	Description
`bool`	Whether the user is an admin, or the deployment lets any
`bool`	authenticated user add types (`custom.access_control.service_types`
`bool`	is `open` or `moderated`; cf.
`bool`	`cloud_registry.ga4gh.registry.service_types`).

service_types_pending ¶

service_types_pending() -> bool

Check whether service types that the current user adds are pending.

Must be called in a request context.

Returns:

Type	Description
`bool`	Whether the deployment moderates service types
`bool`	(`custom.access_control.service_types` is `moderated`) and the user
`bool`	is not an admin, so that their new types are pending until an admin
`bool`	approves them.

current_user ¶

current_user() -> str

Get the subject of the current request, for log messages.

Returns:

Type	Description
`str`	User identifier, with non-printable characters escaped.

remember_current_user ¶

remember_current_user(service_ids: list[str]) -> None

Store the current subject's display name for services with a role.

Stored for the services the subject of the current request has a role on. Names are shown to owners and admins listing the roles on a service, so they are only stored per service, once the user has used the registry with a role on it (and are removed with the role). Nothing is stored if the identity provider reports no name.

Parameters:

Name	Type	Description	Default
`service_ids`	`list[str]`	Identifiers of services the user has a role on.	required

add_owner ¶

add_owner(serviceId: str) -> None

Make the subject of the current request owner of a new service.

See FOCA's add_owner(), which logs the new role.

Parameters:

Name	Type	Description	Default
`serviceId`	`str`	Service identifier.	required

Raises:

Type	Description
`Unauthorized`	The current request has no subject.
`ValueError`	The role can't be stored (e.g., for service identifiers with commas).

remove_owner ¶

remove_owner(serviceId: str) -> None

Revoke the owner role of the current subject on a service.

Also deletes the name stored for the subject of the current request on the service (e.g., if the service they registered was deleted meanwhile). Other roles on the service (e.g., of the owner of a service registered again with the identifier) are kept.

Parameters:

Name	Type	Description	Default
`serviceId`	`str`	Service identifier.	required

Raises:

Type	Description
`Unauthorized`	The current request has no subject.

remove_permissions ¶

remove_permissions(serviceId: str) -> None

Revoke all roles granted on a service.

Parameters:

Name	Type	Description	Default
`serviceId`	`str`	Service identifier.	required

getServicePermissions ¶

getServicePermissions(serviceId: str, **kwargs: Any) -> list[dict[str, Any]]

List roles granted on a service.

Parameters:

Name	Type	Description	Default
`serviceId`	`str`	Service identifier.	required
`**kwargs`	`Any`	Further arguments that Connexion passes (`user`, `token_info`); unused.	`{}`

Returns:

Type	Description
`list[dict[str, Any]]`	Roles granted on the service: user identifier, display name (if
`list[dict[str, Any]]`	known) and role of each role granted to a user, followed by group
`list[dict[str, Any]]`	and role of each role granted to a group.

check_namespaces ¶

check_namespaces() -> None

Check that no trusted issuer has a namespace with the segment groups.

Must be called in an app context. The identifiers of its users would start with groups/, the path of the roles of groups, so their roles couldn't be changed via the permissions API (cf. _check_user_id()).

Raises:

Type	Description
`ValueError`	The namespace of a trusted issuer (`security.auth.issuers`) is `groups` or starts with `groups/`.

warn_owner_rules ¶

warn_owner_rules() -> None

Warn if owners can change roles of groups although that rule is changed.

Deployments that change or remove the rule of owners' roles of groups (OWNER_GROUP_RULE) usually restrict them, but other rules may still allow them: owners-change-permissions (/services/:id/permissions/*) also covers the roles of groups (unlike in 0.5.0). So, unless the rule is kept as in the packaged base config (BASE_CONFIG, which deployments change with overlays), warns of configured rules (named or a list) for owners or any user (*) that allow PUT or DELETE on the roles of both users (EXAMPLE_USERS) and groups (paths of the base rule with example values), i.e., that cover groups as a side effect; rules for groups only (e.g., the changed rule itself) are the deployment's choice. Rules stored via FOCA's permissions API are not checked.

Must be called in an app context.

putServicePermission ¶

putServicePermission(serviceId: str, userId: str, **kwargs: Any) -> dict[str, Any]

Grant a user a role on a service, replacing any other role.

See _put_role().

Parameters:

Name	Type	Description	Default
`serviceId`	`str`	Service identifier.	required
`userId`	`str`	User identifier (`sub` claim of the user's tokens).	required
`**kwargs`	`Any`	Further arguments that Connexion passes (`user`, `token_info`, and the request body); unused.	`{}`

Returns:

Type	Description
`dict[str, Any]`	User identifier and granted role.

Raises:

Type	Description
`NotFound`	The user identifier is `groups` (cf. `_check_user_id()`).

deleteServicePermission ¶

deleteServicePermission(serviceId: str, userId: str, **kwargs: Any) -> str

Revoke all roles of a user on a service (cf. _delete_roles()).

Parameters:

Name	Type	Description	Default
`serviceId`	`str`	Service identifier.	required
`userId`	`str`	User identifier.	required
`**kwargs`	`Any`	Further arguments that Connexion passes (`user`, `token_info`); unused.	`{}`

Returns:

Type	Description
`str`	User identifier.

Raises:

Type	Description
`NotFound`	The user identifier is `groups` (cf. `_check_user_id()`).

putServiceGroupPermission ¶

putServiceGroupPermission(serviceId: str, group: str, **kwargs: Any) -> dict[str, Any]

Grant a group a role on a service, replacing any other role.

See _put_role(). Owners may only grant roles to groups they are members of (i.e., whose group claim contains it), as they could otherwise hand a service to any group; admins may grant roles to any group.

Parameters:

Name	Type	Description	Default
`serviceId`	`str`	Service identifier.	required
`group`	`str`	Group, i.e., a value of the group claim of its members.	required
`**kwargs`	`Any`	Further arguments that Connexion passes (`user`, `token_info`, and the request body); unused.	`{}`

Returns:

Type	Description
`dict[str, Any]`	Group and granted role.

Raises:

Type	Description
`Forbidden`	The subject of the current request is neither a member of the group nor an admin.

deleteServiceGroupPermission ¶

deleteServiceGroupPermission(serviceId: str, group: str, **kwargs: Any) -> str

Revoke all roles of a group on a service (cf. _delete_roles()).

Any owner may revoke roles of groups, also of groups they are not a member of.

Parameters:

Name	Type	Description	Default
`serviceId`	`str`	Service identifier.	required
`group`	`str`	Group.	required
`**kwargs`	`Any`	Further arguments that Connexion passes (`user`, `token_info`); unused.	`{}`

Returns:

Type	Description
`str`	The group.

getMe ¶

getMe(**kwargs: Any) -> dict[str, Any]

Get the current user, and store or refresh their display name.

The name is stored for the services the user has a role on.

Parameters:

Name	Type	Description	Default
`**kwargs`	`Any`	Further arguments that Connexion passes (`user`, `token_info`); unused.	`{}`

Returns:

Type	Description
`dict[str, Any]`	User identifier (`user_id`), whether the user is an admin on this
`dict[str, Any]`	request (`admin`; cf. FOCA's `is_admin()`; in admin mode, i.e., if
`dict[str, Any]`	the request asks for the opt-in role `admin`), whether they may
`dict[str, Any]`	switch to admin mode (`admin_available`; cf. FOCA's
`dict[str, Any]`	`get_available_opt_in_roles()`; always if `admin` is true, e.g.,
`dict[str, Any]`	without opt-in), whether they may add service types
`dict[str, Any]`	(`may_add_service_types`; cf. `may_add_service_types()`) and
`dict[str, Any]`	whether these are pending (`service_types_pending`; cf.
`dict[str, Any]`	`service_types_pending()`), and the user's roles on services (cf.
`dict[str, Any]`	FOCA's `list_subject_roles()`): granted to the user, and granted to
`dict[str, Any]`	groups the user is a member of (with `group`), sorted by service
`dict[str, Any]`	identifier (roles of the user first). Names are only stored for
`dict[str, Any]`	services with roles granted to the user, as only these are listed
`dict[str, Any]`	with names (cf. `getServicePermissions()`).

Raises:

Type	Description
`Unauthorized`	The request has no subject (the endpoint requires authentication, so it always has one).

Migrations¶

namespace_subjects ¶

Namespace the stored subjects of a token issuer, for several issuers.

With several trusted token issuers, FOCA namespaces subjects by issuer (<namespace>|<sub>; by default, the namespace is the issuer), so that equal subjects of different issuers are different users. Subjects stored while the issuer's subjects were bare (e.g., while it was the only trusted issuer) then no longer match its users. This migration rewrites them to the namespaced form:

users in stored policy rules (user:<sub>, e.g., owners and maintainers of services, or roles granted via FOCA's permissions API); configured rules (security.access_control.policies) are left as they are, as the app replaces them with those of the config on startup, so change the config instead;
the users who added service types (created_by of service_types), for the limit of unused types and the visibility of pending types;
the users of the stored display names (user_id of user_names).

Subjects that are already namespaced (with |) are kept, so the migration can be run again, e.g., after an interrupted run. It then marks the stored rules as changed, so that the cached enforcers of running app instances load them again (FOCA's Adapter.mark_changed()).

Run it once the config trusts the new issuer (so that it names the namespace), with the app's config (FOCA_CONFIG_OVERLAY), e.g., in a container of the app: python -m cloud_registry.namespace_subjects <namespace> (add --dry-run to only count the subjects). Until then, the issuer's stored users lose their roles. Prefer namespace: "" for the issuer trusted so far, which keeps its subjects bare (see "Several token issuers" in docs/upgrading.md).

check_namespace ¶

check_namespace(namespace: str) -> None

Check that a namespace is one of the trusted issuers.

Must be called in an app context.

Parameters:

Name	Type	Description	Default
`namespace`	`str`	Namespace of the issuer's subjects.	required

Raises:

Type	Description
`ValueError`	The namespace is empty or invalid (cf. FOCA's `IssuerConfig`), no trusted issuer has it (`security.auth.issuers`), e.g., as the config doesn't trust the new issuer yet, or a trusted issuer keeps its subjects bare: the bare subjects are then that issuer's, also those stored later.

namespace_subjects ¶

namespace_subjects(namespace: str, dry_run: bool = False) -> dict[str, int]

Namespace the stored subjects of an issuer.

Must be called in an app context. All bare subjects are taken as the issuer's: run it for the issuer whose subjects were stored bare, i.e., the one trusted so far.

Parameters:

Name	Type	Description	Default
`namespace`	`str`	Namespace of the issuer's subjects, as in the config (by default, the issuer, e.g., `https://login.example.org/`).	required
`dry_run`	`bool`	Only count the subjects to rewrite, without changes.	`False`

Returns:

Type	Description
`dict[str, int]`	Numbers of stored policy rules (`rules`), service types
`dict[str, int]`	(`service_types`) and display names (`user_names`) of bare users
`dict[str, int]`	(rewritten, unless `dry_run`), and of configured rules with bare
`dict[str, int]`	users (`configured_rules`; kept, cf. the module docs).

Raises:

Type	Description
`ValueError`	No trusted issuer has the namespace (cf. `check_namespace()`).

main ¶

main(argv: Sequence[str] | None = None) -> int

Namespace the stored subjects of an issuer, with the app's config.

Parameters:

Name	Type	Description	Default
`argv`	`Sequence[str] \| None`	Command line arguments, without the program name: the namespace, and optionally `--dry-run`.	`None`

Returns:

Type	Description
`int`	The exit status: 0 on success, 1 if the namespace is not that of a
`int`	trusted issuer.

Errors¶

exceptions ¶

App exceptions.

CodedConflict ¶

Bases: Conflict

409 Conflict with a stable code, for clients to tell conflicts apart.

The code is served as the extension member code of the problem details built with problem(), next to the unchanged detail, and documented in the API specification (schema ConflictError). Conflicts that reach FOCA's error handler instead (i.e., raised, not returned) get the code of their entry in exceptions (default: CONFLICT_CODE), so add an entry for those, as for FOCA's LastOwner.

Attributes:

Name	Type	Description
`problem_code`	`str`	Code of the conflict (kebab case); not `code`, which is the status code of werkzeug's exceptions.

ServiceBusy ¶

ServiceBusy()

Bases: CodedConflict

The service is being renamed, or changed while it is to be renamed.

Raised by changes of services that waited in vain for a rename to finish, and returned by renames that waited in vain for changes or another rename (cf. cloud_registry.ga4gh.registry.locks); in exceptions, so that the raised one has its code, too.

Create the exception, with a description of what to do.

problem ¶

problem(exc: HTTPException, **extensions: Any) -> Response

Build a problem details response with the description of an exception.

See FOCA's problem_response(); conflicts (409) have their code (code): the one of the class for coded conflicts (cf. CodedConflict), otherwise the one of their entry in exceptions.

Parameters:

Name	Type	Description	Default
`exc`	`HTTPException`	werkzeug exception with a request-specific description.	required
`extensions`	`Any`	Extension members of the problem details; for conflicts, not `code`, which is set from the exception.	`{}`

Returns:

Type	Description
`Response`	Problem details response.