Security

Security and authentication utilities

Secure connectivity management for NiPyApi

nipyapi.security.add_user_group_to_access_policy(user_group, policy, service='nifi', refresh=True)[source]

Attempts to add the given user group object to the given access policy

Parameters:
Returns:

Updated Policy object

nipyapi.security.add_user_to_access_policy(user, policy, service='nifi', refresh=True, strict=True)[source]

Attempts to add the given user object to the given access policy

Parameters:

  • user (User) or (UserEntity) – User object to add

  • policy (AccessPolicyEntity) or (AccessPolicy) – Access Policy object

  • service (str) – ‘nifi’ or ‘registry’ to identify the target service

  • refresh (bool) – Whether to refresh the policy object before submit

  • strict (bool) – If True, will return error if user already present, if False will ignore the already exists

Returns:

Updated Policy object

nipyapi.security.bootstrap_security_policies(service, user_identity=None, group_identity=None, nifi_proxy_identity=None)[source]

Creates a default security context within NiFi or Nifi-Registry.

Parameters:

  • service (str) – The service to configure security for (‘nifi’ or ‘registry’)

  • user_identity (nipyapi.nifi.UserEntity or nipyapi.registry.User, optional) – User identity to apply policies to

  • group_identity (nipyapi.nifi.UserGroupEntity or nipyapi.registry.UserGroup, optional) – Group identity to apply policies to

Returns:

None

nipyapi.security.create_access_policy(resource, action, r_id=None, service='nifi')[source]

Creates an access policy for the given resource, action and optionally resource_id for NiFi.

Parameters:

  • resource (str) – a valid resource type for this service, e.g. ‘bucket’

  • action (str) – a valid action type for this service, typically ‘read’, ‘write’ or ‘delete’

  • r_id (optional[str]) – if NiFi, the resource ID of the resource

  • service (str) – the service to target

Returns:

An access policy object for that service

nipyapi.security.create_service_user(identity, service='nifi', strict=True)[source]

Attempts to create a user with the provided identity in the given service

Parameters:

  • identity (str) – Identity string for the user

  • service (str) – ‘nifi’ or ‘registry’

  • strict (bool) – If Strict, will error if user already exists

Returns:

The new User or UserEntity object

nipyapi.security.create_service_user_group(identity, service='nifi', users=None, strict=True)[source]

Attempts to create a user with the provided identity and member users in the given service

Parameters:

  • identity (str) – Identity string for the user group

  • service (str) – ‘nifi’ or ‘registry’

  • users (list) – A list of nifi.UserEntity or registry.User belonging to the group

  • strict (bool) – Whether to throw an error on already exists

Returns:

The new UserGroup or UserGroupEntity object

nipyapi.security.create_ssl_context_controller_service(parent_pg, name, keystore_file=None, keystore_password=None, truststore_file=None, truststore_password=None, key_password=None, keystore_type=None, truststore_type=None, ssl_protocol=None, ssl_service_type=None)[source]

Creates and configures an SSL Context Service for secure client connections. Note that once created it can be listed and deleted using the standard canvas functions.

Parameters:

  • parent_pg (ProcessGroupEntity) – The Process Group to create the service in

  • name (str) – Name for the SSL Context Service

  • keystore_file (str) – Path to the client certificate/keystore file

  • keystore_password (str) – Password for the keystore

  • truststore_file (str) – Path to the truststore file

  • truststore_password (str) – Password for the truststore

  • key_password (Optional[str]) – Password for the key, defaults to keystore_password if not set

  • keystore_type (Optional[str]) – Type of keystore (JKS, PKCS12), defaults to JKS

  • truststore_type (Optional[str]) – Type of truststore (JKS, PKCS12), defaults to JKS

  • ssl_protocol (Optional[str]) – SSL protocol version, defaults to TLS

  • ssl_service_type (Optional[str]) – SSL service type, defaults to StandardRestrictedSSLContextService

Returns:

The configured SSL Context Service

Return type:

(ControllerServiceEntity)

nipyapi.security.ensure_ssl_context(name, parent_pg, keystore_file=None, keystore_password=None, truststore_file=None, truststore_password=None, key_password=None, keystore_type=None, truststore_type=None, ssl_protocol=None, ssl_service_type=None)[source]

Ensures an SSL Context Service exists, creating it if necessary.

This is a convenience function that implements the common pattern of: 1. Try to get existing SSL context service by name 2. If not found, create it 3. Ensure it’s scheduled/enabled 4. Handle race conditions gracefully

Parameters:

  • name (str) – Name for the SSL Context Service

  • parent_pg (ProcessGroupEntity) – The Process Group to create the service in

  • keystore_file (str) – Path to the client certificate/keystore file

  • keystore_password (str) – Password for the keystore

  • truststore_file (str) – Path to the truststore file

  • truststore_password (str) – Password for the truststore

  • key_password (Optional[str]) – Password for the key, defaults to keystore_password if not set

  • keystore_type (Optional[str]) – Type of keystore (JKS, PKCS12), defaults to JKS

  • truststore_type (Optional[str]) – Type of truststore (JKS, PKCS12), defaults to JKS

  • ssl_protocol (Optional[str]) – SSL protocol version, defaults to TLS

  • ssl_service_type (Optional[str]) – SSL service type, defaults to StandardRestrictedSSLContextService

Returns:

The SSL context service (existing or new)

Return type:

(ControllerServiceEntity)

nipyapi.security.get_access_policy_for_resource(resource, action, r_id=None, service='nifi', auto_create=False)[source]

Attempts to retrieve the access policy for a given resource and action, and optionally resource_id if targeting NiFi. Optionally creates the policy if it doesn’t already exist

Parameters:

  • resource (str) – A valid resource in the target service

  • action (str) – A valid action, typically ‘read’, ‘write’ or ‘delete’

  • r_id (Optional[str]) – The UUID of the resource, valid only if targeting NiFi resources

  • service (str) – Which service to target, typically ‘nifi’ or ‘registry’

  • auto_create (bool) – Whether to create the targeted policy if it doesn’t already exist

Returns:

The relevant AccessPolicy object

nipyapi.security.get_service_access_status(service='nifi', bool_response=False)[source]

Gets the access status for the current session

Parameters:

  • service (str) – A String of ‘nifi’ or ‘registry’ to indicate which service to check status for

  • bool_response (bool) – If True, the function will return False on hitting an Error instead of raising it. Useful for connection testing.

Returns:

(bool) if bool_response, else the Service Access Status of the User

nipyapi.security.get_service_user(identifier, identifier_type='identity', service='nifi')[source]

Gets the unique user matching to the given identifier and type.

Parameters:

  • identifier (str) – the string to search for

  • identifier_type (str) – the field to search in

  • service (str) – the name of the service

Returns:

None if no match, else single object

nipyapi.security.get_service_user_group(identifier, identifier_type='identity', service='nifi')[source]

Gets the unique group matching to the given identifier and type.

Parameters:

  • identifier (str) – the string to search for

  • identifier_type (str) – the field to search in, identity or id

  • service (str) – the name of the service

Returns:

None if no match, else single object

nipyapi.security.list_service_user_groups(service='nifi')[source]

Returns list of service user groups for a given service :param service: ‘nifi’ or ‘registry’ :type service: str

Returns:

[(nifi.UserGroupEntity, registry.UserGroup)]

nipyapi.security.list_service_users(service='nifi')[source]

Lists all users of a given service, takes a service name as a string

nipyapi.security.remove_service_user(user, service='nifi', strict=True)[source]

Removes a given User from the given Service

Parameters:

  • user – [(nifi.UserEntity), (registry.User)] Target User object

  • service (str) – ‘nifi’ or ‘registry’

  • strict (bool) – Whether to throw an error if User not found

Returns:

Updated User Entity or None

nipyapi.security.remove_service_user_group(group, service='nifi', strict=True)[source]

Removes a given User Group from the given Service

Parameters:

  • group – [(nifi.UserEntity), (registry.User)] Target User object

  • service (str) – ‘nifi’ or ‘registry’

  • strict (bool) – Whether to throw an error if User not found

Returns:

Updated User Group or None

nipyapi.security.reset_service_connections(service=None)[source]

Reset service connections by logging out and clearing API clients.

Parameters:

service (str, optional) – ‘nifi’, ‘registry’, or None for both services. Defaults to None (both services).

nipyapi.security.service_login(service='nifi', username=None, password=None, bool_response=False)[source]

Login to the currently configured NiFi or NiFi-Registry server.

Login requires a secure connection over https. Prior to calling this method, the host must be specified and the SSLContext should be configured (if necessary).

Successful login will result in a generated token (JWT) being cached in the api_client config that will be passed in all future REST API calls. To clear that token, call service_logout.

The token is temporary and will expire after a duration set by the server. After a token expires, you must call this method again to generate a new token.

Parameters:

  • service (str) – ‘nifi’ or ‘registry’; the service to login to

  • username (str) – The username to submit

  • password (str) – The password to use

  • bool_response (bool) – If True, the function will return False instead of an error. Useful for connection testing.

Returns:

True if successful, False or an Error if not. See bool_response

Return type:

(bool)

nipyapi.security.service_login_oidc(service='nifi', username=None, password=None, oidc_token_endpoint=None, client_id=None, client_secret=None, bool_response=False, return_token_info=False, verify_ssl=None)[source]

Login to NiFi using OpenID Connect (OIDC) OAuth2 authentication.

This method acquires an access token from an OIDC provider using either: - OAuth2 Resource Owner Password Credentials flow (RFC 6749) - when username/password provided - OAuth2 Client Credentials flow (RFC 6749) - when only client_id/client_secret provided

The method auto-detects which flow to use based on the provided parameters and configures the service to use bearer token authentication.

Parameters:

  • service (str) – ‘nifi’ or ‘registry’; the service to login to (currently only ‘nifi’ supports OIDC)

  • username (str, optional) – The username for Resource Owner Password flow. If provided, password must also be provided.

  • password (str, optional) – The password for Resource Owner Password flow. If provided, username must also be provided.

  • oidc_token_endpoint (str) – The OIDC token endpoint URL (e.g., ‘http://localhost:8080/realms/nipyapi/protocol/openid-connect/token’)

  • client_id (str) – The OIDC client ID

  • client_secret (str) – The OIDC client secret

  • bool_response (bool) – If True, the function will return False instead of an error. Useful for connection testing.

  • return_token_info (bool) – If True, return the full OAuth2 token response instead of just a boolean. Useful for extracting user identity information.

  • verify_ssl (bool, optional) – Whether to verify SSL certificates for the OIDC endpoint. If None (default), uses the service’s SSL verification setting. Set to False for development environments with self-signed certificates.

Returns:

True if successful (default), False if bool_response=True and failed,

or full OAuth2 token response dict if return_token_info=True. See bool_response and return_token_info.

Return type:

(bool or dict)

Raises:

ValueError – If required parameters are missing or invalid combination provided

nipyapi.security.service_logout(service='nifi')[source]

Logs out from the service by resetting the token :param service: ‘nifi’ or ‘registry’; the target service :type service: str

Returns:

True of access removed, False if still set

Return type:

(bool)

nipyapi.security.set_service_auth_token(token=None, token_name='bearerAuth', service='nifi')[source]

Helper method to set the auth token correctly for the specified service

Parameters:

  • token (Optional[str]) – The token to set. Defaults to None.

  • token_name (str) – the api_key field name to set the token to. Defaults to ‘bearerAuth’

  • service (str) – ‘nifi’ or ‘registry’, the service to set

Returns:

True on success, False if token not set

Return type:

(bool)

nipyapi.security.set_service_ssl_context(service='nifi', ca_file=None, client_cert_file=None, client_key_file=None, client_key_password=None, purpose=None)[source]

Create an SSLContext for connecting over https to a secured NiFi or NiFi-Registry instance.

This method can be used to create an SSLContext for two-way TLS in which a client cert is used by the service to authenticate the client.

This method can also be used for one-way TLS in which the client verifies the server’s certificate, but authenticates using a different form of credentials, such as LDAP username/password.

If you are using one-way TLS with a certificate signed by a root CA trusted by your system/platform, this step is not necessary as the default TLS-handshake should “just work.”

Parameters:

  • service (str) – ‘nifi’ or ‘registry’ to indicate which service config to set the ssl context to

  • ca_file (str) – A PEM file containing certs for the root CA(s) for the NiFi Registry server

  • client_cert_file (str) – A PEM file containing the public certificates for the user/client identity

  • client_key_file (str) – An encrypted (password-protected) PEM file containing the client’s secret key

  • client_key_password (str) – The password to decrypt the client_key_file

  • purpose (ssl.Purpose) – The purpose of the SSLContext

Returns:

(None)

nipyapi.security.set_shared_ca_cert(ca_cert_path)[source]

Set CA certificate for both NiFi and Registry services.

This is the typical pattern since both services usually trust the same Certificate Authority.

Parameters:

ca_cert_path (str) – Path to the CA certificate file

nipyapi.security.set_ssl_warning_suppression(suppress_warnings)[source]

Control SSL warning suppression.

Parameters:

suppress_warnings (bool) – Whether to suppress SSL warnings. True = suppress warnings (development/self-signed certs) False = show warnings (production/security audit)

nipyapi.security.update_access_policy(policy, service='nifi')[source]

Applies an updated access policy to the service indicated

Parameters:

  • policy (PolicyEntity) – The policy object to submit

  • service (str) – ‘nifi’ or ‘registry’ to indicate the target service

Returns:

The updated policy if successful

Return type:

AccessPolicyEntity