hipcheck_sdk package

Subpackages

Submodules

hipcheck_sdk.cli module

hipcheck_sdk.cli.get_parser_for(plugin_name: str) → ArgumentParser

Get the default argument parser for a Hipcheck plugin

Returns:

An ArgumentParser configured to capture Hipcheck plugin CLI arguments

hipcheck_sdk.cli.run_server_for(plugin: Plugin)

Parse CLI arguments and start the server for the plugin. Does not return until the gRPC connection closes.

Parameters:

plugin (Plugin) – An instance of a subclass of Plugin

hipcheck_sdk.engine module

class hipcheck_sdk.engine.PluginEngine(session_id: int = 0, tx: Queue | None = None, rx: Queue | None = None, drop_tx: Queue | None = None, mock_responses: List[Tuple[Tuple[str, object], object]] | None = None)

Bases: object

Manages a particular query session.

An instance of this class invokes a query endpoint, passing a handle to itself. This allows the query endpoint to request information from other Hipcheck plugins as part of its logic.

batch(target: str) → QueryBuilder

Create a QueryBuilder instance to dynamically aggregate keys for a batch query against target as opposed to using PluginEngine.batch_query() which requires having all keys available immediately

Parameters:

target (str) – A string of the form <PUBLISHER>/<PLUGIN>/<ENDPOINT>, where /<ENDPOINT> may be omitted if targeting a default endpoint. Indicates the remote plugin endpoint to query.

Returns:

An instance of QueryBuilder

async batch_query(target: str, keys: List[object]) → List[object]

Query another Hipcheck plugin endpoint target with a list of keys

Parameters:

• target (str) – Indicates the remote plugin endpoint to query. Has the same format requirements as query()

• keys (list) – The list of keys to send as a single batch query to Hipcheck core.

Returns:

A list of output values corresponding to each element of key being applied to the target endpoint.

Raises:

SdkError

mock()

For unit testing purposes, construct a PluginEngine with a set of mock responses

Parameters:

mock_responses (tuple) – A list of key, value pairs that maps queries to mock responses. Does not use a dict since many relevant types are not hashable and thus cannot be used as keys. The query is a tuple of a target string and a key object, the response is the output object for that query.

Returns:

An instance of PluginEngine

async query(target: str, key: object) → object

Query another Hipcheck plugin endpoint target with key input

Parameters:

• target (str) – A string of the form <PUBLISHER>/<PLUGIN>/<ENDPOINT>, where /<ENDPOINT> may be omitted if targeting a default endpoint. Indicates the remote plugin endpoint to query.

• key (object) – The key for the query

Returns:

The deserialized result

Raises:

SdkError

async query_inner(target: str, keys: List[object]) → List[object]

Raises:

SdkError

record_concern(concern: str)

Records a concern that will be emitted in the final Hipcheck report. Intended for use within a @query-decorated endpoint function.

Parameters:

concern (str) – The concern to be recorded

class hipcheck_sdk.engine.QueryBuilder(engine, target: str)

Bases: object

An alternative to calling PluginEngine.batch_query(), an instance of this class is returned by PluginQuery.batch(), and allows plugin authors to build up a list of keys for a batch query over time, then call QueryBuilder.send() to send them in a single message to Hipcheck.

query(key: object) → int

Adds a key to the batch query being built by this object

Parameters:

key (object) – The key to add to the query batch.

Returns:

The index of the list at which the key was added. Can be used to index the output of send() to get the corresponding output for that key.

async send() → List[object]

Sends all keys aggregated with query() in a single batch query to Hipcheck core.

Returns:

The list of output objects for the keys aggregated in this object

Raises:

SdkError –

hipcheck_sdk.error module

exception hipcheck_sdk.error.ConfigError

Bases: Exception

Parent class for all errors defined and caught during Plugin.set_config()

exception hipcheck_sdk.error.FailedToSendQueryFromSessionToServer

Bases: SdkError

An error occurred while sending a query message to Hipcheck core

exception hipcheck_sdk.error.InvalidConfigValue(field_name: str, value: str, reason: str)

Bases: ConfigError

The value value of config field field_name was invalid because of reason

field_name: str

reason: str

value: str

exception hipcheck_sdk.error.InvalidJsonInQueryKey

Bases: SdkError

One of the key fields in the query object contained invalid JSON

exception hipcheck_sdk.error.InvalidJsonInQueryOutput

Bases: SdkError

One of the output fields in the query object contained invalid JSON

exception hipcheck_sdk.error.InvalidQueryTargetFormat

Bases: SdkError

The target string for a query was incorrectly formatted

exception hipcheck_sdk.error.MissingRequiredConfig(field_name: str, field_type: str, possible_values: list[str])

Bases: ConfigError

The config field field_name of type field_type was missing. Possible values include possible_values.

field_name: str

field_type: str

possible_values: list[str]

exception hipcheck_sdk.error.MoreAfterQueryComplete(id: int)

Bases: SdkError

The session with the given id received more messages after a complete query was parsed

id: int

exception hipcheck_sdk.error.ReceivedReplyWhenExpectingSubmitChunk

Bases: SdkError

Received a Reply-type query message when a Submit-type was expected

exception hipcheck_sdk.error.ReceivedSubmitWhenExpectingReplyChunk

Bases: SdkError

Received a Submit-type query message when a Reply-type was expected

exception hipcheck_sdk.error.SdkError

Bases: Exception

Parent class for all errors defined and caught by the SDK

exception hipcheck_sdk.error.UnexpectedReplyInProgress

Bases: SdkError

Received a query with the unexpected state ReplyInProgress

exception hipcheck_sdk.error.UnexpectedRequestInProgress

Bases: SdkError

Received a query with the unexpected state RequestInProgress

exception hipcheck_sdk.error.UnknownPluginQuery

Bases: SdkError

The target for a query was unrecognized

exception hipcheck_sdk.error.UnrecognizedConfig(field_name: str, field_value: str, possible_confusables: list[str])

Bases: ConfigError

The config field field_name with value field_value was unrecognized. Intended field name may have possibly been one of possible_confusables.

field_name: str

field_value: str

possible_confusables: list[str]

exception hipcheck_sdk.error.UnspecifiedConfigError(message: str)

Bases: ConfigError

An unspecified error occurred during configuration

message: str

exception hipcheck_sdk.error.UnspecifiedQueryState

Bases: SdkError

Catchall error used when other SdkError variants don’t apply

hipcheck_sdk.options module

class hipcheck_sdk.options.SdkOptions(rfd9_compat: bool = True)

Bases: object

Options that can be set before starting a server.PluginServer to control behavior. Designed to offer roughly similar behavior to the feature flags available in the Rust Hipcheck SDK.

rfd9_compat: bool = True

hipcheck_sdk.options.enabled(option_name: str) → bool

Parameters:

option_name (str) – The option name to check

Returns:

True if the option is enabled, else False

hipcheck_sdk.options.get_options() → SdkOptions

Returns:

A copy of the global SdkOptions instance. Note that this is a deep copy so changes to the returned value will not affect the global state.

hipcheck_sdk.options.set_option(option_name: str, val: bool)

Turn an option on or off.

Parameters:

• option_name (str) – The option name to set

• val (bool) – The value to which to set the option

hipcheck_sdk.options.set_options(options: SdkOptions)

Set the entire global SdkOptions instance at once instead of toggling individual options.

Parameters:

options (SdkOptions) – The options instance to override the current global value with

hipcheck_sdk.query module

hipcheck_sdk.query.query(f_py=None, default: bool = False, key_schema: dict | None = None, output_schema=None)

Decorator function for query endpoints. Endpoint functions must have the following signature:

async fn <QUERY_NAME>(engine: hipcheck_sdk.engine.PluginEngine, key: <TYPE>) -> <TYPE>

Parameters:

• default (bool) – Whether this endpoint is the default for the plugin

• key_schema (dict) – A jsonable dict representing the schema for the key to this enpdpoint. If None, derive from type hint on key parameter of function. If a schema is supplied explicitly instead of having the SDK use the type hint, the object passed to the query func will not be automatically converted to a class instance.

• output_schema (dict) – A jsonable dict representing the schema for the return value of this endpoint. If None, derive from type hint on return value of function.

Raises:

TypeError – The function lacked type hints or a JSON schema could not be derived from them

hipcheck_sdk.server module

class hipcheck_sdk.server.Plugin

Bases: ABC

default_policy_expr() → str | None

Return the plugin’s default policy expression. This will only ever be called after Plugin.set_config(). This should only be overriden if the plugin defines a default query endpoint. For more information on policy expression syntax, see the Hipcheck website.

Returns:

The default policy expression

default_query() → Endpoint | None

Get the plugin’s default query, if it has one. The default query is a @query function with default=True in the decorator arguments.

Returns:

The endpoint instance marked default, if one exists else None

explain_default_query() → str | None

This should only be overriden if the plugin defines a default query endpoint.

Returns:

An unstructured description of what is returned by the plugin’s default query endpoint.

set_config(config: Dict[str, object])

Configure the plugin according to the fields received from the policy file used for this analysis.

Parameters:

config (dict) – The configuration key-value map

Raises:

ConfigError – The config value was invalid

class hipcheck_sdk.server.PluginServer(plugin: Plugin)

Bases: PluginServiceServicer

The server object which runs a plugin class implementation

init_logger(log_level_str=<class 'str'>)

Setup plugin logger in JSON at appropriate level.

Parameters:

log_level_str (str) – maximum produced log level for plugin

listen(port: int, host='127.0.0.1')

Start the plugin listening for an incoming gRPC connection from Hipcheck core

Parameters:

• port (int) – The port on which to listen

• host (str) – The host IP on which to listen. Defaults to loopback, for plugins that will be run in a docker container you will need to change it to listen on all network interfaces, e.g. ‘0.0.0.0’.

register(log_level='error', init_logger=True)

Set the server to use plugin as its implementation

Parameters:

• plugin (Plugin) – The plugin instance with which to run

• log_level (str) – A string indicating the minimum logging level to emit

• init_logger (bool) – If True, init the standard plugin logger that emits a custom JSON format over stderr to Hipcheck core.

Module contents