Module sysbot.connectors.http

HTTP/HTTPS Connector with Multiple Authentication Methods

This module provides a generic HTTP/HTTPS connector with support for various authentication methods. Each authentication method is implemented as a separate self-contained class.

Classes

class Apikey (port=443, use_https=True)

HTTP connector with API Key authentication. Supports API keys in headers or query parameters.

Initialize API Key authentication connector.

Args

port : int
Default port (default: 443).
use_https : bool
Whether to use HTTPS (default: True).

Ancestors

Methods

def close_session(self, session)

Close the session (no-op for API key auth).

Args

session : dict
Session configuration.
def execute_command(self, session, command, options=None)

Execute an HTTP request with API key authentication.

Args

session : dict
Session configuration.
command : str
API endpoint path.
options : dict
Optional request parameters: - method (str): HTTP method (default: GET) - params (dict): URL query parameters - headers (dict): HTTP headers - data: Request body data - json: JSON request body - verify (bool): Verify SSL certificates (default: True)

Returns

bytes
Response content.
def open_session(self,
host,
port=None,
login=None,
password=None,
api_key=None,
api_key_header='X-API-Key',
api_key_in_query=False,
verify_ssl=True)

Opens a session with API key authentication.

Args

host : str
Hostname or IP address.
port : int
Port. If None, uses default_port.
login : str
Not used for API key auth (for compatibility).
password : str
Not used for API key auth (for compatibility).
api_key : str
The API key.
api_key_header : str
Header name for API key (default: "X-API-Key").
api_key_in_query : bool
If True, send API key as query parameter instead of header.
verify_ssl : bool
Whether to verify SSL certificates (default: True). Set to False to disable SSL verification for self-signed certificates.

Returns

dict
Session configuration.
class BaseHttp (port=443, use_https=True)

Base class for HTTP/HTTPS connectors providing common functionality. This class should not be used directly but extended by authentication-specific classes.

Initialize base HTTP connector.

Args

port : int
Default port (default: 443 for HTTPS, 80 for HTTP).
use_https : bool
Whether to use HTTPS (default: True).

Ancestors

Subclasses

class Basicauth (port=443, use_https=True)

HTTP connector with Basic Authentication (username/password).

Initialize Basic Auth connector.

Args

port : int
Default port (default: 443).
use_https : bool
Whether to use HTTPS (default: True).

Ancestors

Methods

def close_session(self, session)

Close the session (no-op for Basic auth).

Args

session : dict
Session configuration.
def execute_command(self, session, command, options=None)

Execute an HTTP request with Basic authentication.

Args

session : dict
Session configuration.
command : str
API endpoint path.
options : dict
Optional request parameters: - method (str): HTTP method (default: GET) - params (dict): URL query parameters - headers (dict): HTTP headers - data: Request body data - json: JSON request body - verify (bool): Verify SSL certificates (default: True)

Returns

bytes
Response content.
def open_session(self, host, port=None, login=None, password=None, verify_ssl=True)

Opens a session with Basic authentication.

Args

host : str
Hostname or IP address.
port : int
Port. If None, uses default_port.
login : str
Username.
password : str
Password.
verify_ssl : bool
Whether to verify SSL certificates (default: True). Set to False to disable SSL verification for self-signed certificates.

Returns

dict
Session configuration.
class Certificate (port=443, use_https=True)

HTTP connector with Client Certificate authentication (mutual TLS).

Initialize Certificate connector.

Args

port : int
Default port (default: 443).
use_https : bool
Whether to use HTTPS (default: True).

Ancestors

Methods

def close_session(self, session)

Close the session (no-op for certificate auth).

Args

session : dict
Session configuration.
def execute_command(self, session, command, options=None)

Execute an HTTP request with certificate authentication.

Args

session : dict
Session configuration.
command : str
API endpoint path.
options : dict
Optional request parameters (method, params, headers, data, json).

Returns

bytes
Response content.
def open_session(self,
host,
port=None,
login=None,
password=None,
cert_file=None,
key_file=None,
ca_bundle=None,
verify_ssl=True)

Opens a session with client certificate authentication.

Args

host : str
Hostname or IP address.
port : int
Port. If None, uses default_port.
login : str
Not used (for compatibility).
password : str
Password for encrypted key file (optional).
cert_file : str
Path to client certificate file.
key_file : str
Path to private key file (optional, can be in cert_file).
ca_bundle : str
Path to CA bundle for server verification (optional).
verify_ssl : bool
Whether to verify SSL certificates (default: True). Set to False to disable SSL verification for self-signed certificates. If ca_bundle is provided, it takes precedence over verify_ssl.

Returns

dict
Session configuration.
class Hmac (port=443, use_https=True)

HTTP connector with HMAC authentication.

Initialize HMAC connector.

Args

port : int
Default port (default: 443).
use_https : bool
Whether to use HTTPS (default: True).

Ancestors

Methods

def close_session(self, session)

Close the session (no-op for HMAC).

Args

session : dict
Session configuration.
def execute_command(self, session, command, options=None)

Execute an HTTP request with HMAC authentication.

Args

session : dict
Session configuration.
command : str
API endpoint path.
options : dict
Optional request parameters (method, params, headers, data, json).

Returns

bytes
Response content.
def open_session(self,
host,
port=None,
login=None,
password=None,
secret_key=None,
algorithm='sha256',
signature_header='X-Signature',
timestamp_header='X-Timestamp',
verify_ssl=True)

Opens a session with HMAC authentication.

Args

host : str
Hostname or IP address.
port : int
Port. If None, uses default_port.
login : str
Access key/client ID.
password : str
Not used (for compatibility).
secret_key : str
Secret key for HMAC signing.
algorithm : str
Hash algorithm (sha256, sha1, sha512, etc.).
signature_header : str
Header name for signature (default: X-Signature).
timestamp_header : str
Header name for timestamp (default: X-Timestamp).
verify_ssl : bool
Whether to verify SSL certificates (default: True). Set to False to disable SSL verification for self-signed certificates.

Returns

dict
Session configuration.
class Jwt (port=443, use_https=True)

HTTP connector with JWT (JSON Web Token) authentication.

Initialize JWT connector.

Args

port : int
Default port (default: 443).
use_https : bool
Whether to use HTTPS (default: True).

Ancestors

Methods

def close_session(self, session)

Close the session (no-op for JWT).

Args

session : dict
Session configuration.
def execute_command(self, session, command, options=None)

Execute an HTTP request with JWT authentication.

Args

session : dict
Session configuration.
command : str
API endpoint path.
options : dict
Optional request parameters (method, params, headers, data, json).

Returns

bytes
Response content.
def open_session(self,
host,
port=None,
login=None,
password=None,
secret_key=None,
algorithm='HS256',
token=None,
payload=None,
expiration_minutes=60,
verify_ssl=True)

Opens a session with JWT authentication.

Args

host : str
Hostname or IP address.
port : int
Port. If None, uses default_port.
login : str
Username to include in JWT payload (optional).
password : str
Not used (for compatibility).
secret_key : str
Secret key for signing JWT.
algorithm : str
JWT signing algorithm (default: HS256).
token : str
Existing JWT token (optional).
payload : dict
Custom JWT payload (optional).
expiration_minutes : int
Token expiration time in minutes (default: 60).
verify_ssl : bool
Whether to verify SSL certificates (default: True). Set to False to disable SSL verification for self-signed certificates.

Returns

dict
Session configuration.
class Oauth1 (port=443, use_https=True)

HTTP connector with OAuth 1.0 authentication.

Initialize OAuth 1.0 connector.

Args

port : int
Default port (default: 443).
use_https : bool
Whether to use HTTPS (default: True).

Ancestors

Methods

def close_session(self, session)

Close the session (no-op for OAuth 1.0).

Args

session : dict
Session configuration.
def execute_command(self, session, command, options=None)

Execute an HTTP request with OAuth 1.0 authentication.

Args

session : dict
Session configuration.
command : str
API endpoint path.
options : dict
Optional request parameters: - method (str): HTTP method (default: GET) - params (dict): URL query parameters - headers (dict): HTTP headers - data: Request body data - json: JSON request body - verify (bool): Verify SSL certificates (default: True)

Returns

bytes
Response content.
def open_session(self,
host,
port=None,
login=None,
password=None,
client_key=None,
client_secret=None,
resource_owner_key=None,
resource_owner_secret=None,
verify_ssl=True)

Opens a session with OAuth 1.0 authentication.

Args

host : str
Hostname or IP address.
port : int
Port. If None, uses default_port.
login : str
Not used (for compatibility).
password : str
Not used (for compatibility).
client_key : str
OAuth consumer key.
client_secret : str
OAuth consumer secret.
resource_owner_key : str
OAuth token.
resource_owner_secret : str
OAuth token secret.
verify_ssl : bool
Whether to verify SSL certificates (default: True). Set to False to disable SSL verification for self-signed certificates.

Returns

dict
Session configuration.
class Oauth2 (port=443, use_https=True)

HTTP connector with OAuth 2.0 authentication.

Initialize OAuth 2.0 connector.

Args

port : int
Default port (default: 443).
use_https : bool
Whether to use HTTPS (default: True).

Ancestors

Methods

def close_session(self, session)

Close the session (no-op for OAuth 2.0).

Args

session : dict
Session configuration.
def execute_command(self, session, command, options=None)

Execute an HTTP request with OAuth 2.0 authentication.

Args

session : dict
Session configuration.
command : str
API endpoint path.
options : dict
Optional request parameters: - method (str): HTTP method (default: GET) - params (dict): URL query parameters - headers (dict): HTTP headers - data: Request body data - json: JSON request body - verify (bool): Verify SSL certificates (default: True)

Returns

bytes
Response content.
def open_session(self,
host,
port=None,
login=None,
password=None,
client_id=None,
client_secret=None,
token_url=None,
access_token=None,
refresh_token=None,
verify_ssl=True)

Opens a session with OAuth 2.0 authentication.

Args

host : str
Hostname or IP address.
port : int
Port. If None, uses default_port.
login : str
Not used (for compatibility).
password : str
Not used (for compatibility).
client_id : str
OAuth 2.0 client ID.
client_secret : str
OAuth 2.0 client secret.
token_url : str
URL to obtain tokens.
access_token : str
Existing access token (optional).
refresh_token : str
Refresh token (optional).
verify_ssl : bool
Whether to verify SSL certificates (default: True). Set to False to disable SSL verification for self-signed certificates.

Returns

dict
Session configuration.
class Openidconnect (port=443, use_https=True)

HTTP connector with OpenID Connect authentication.

Initialize OpenID Connect connector.

Args

port : int
Default port (default: 443).
use_https : bool
Whether to use HTTPS (default: True).

Ancestors

Methods

def close_session(self, session)

Close the session (no-op for OpenID Connect).

Args

session : dict
Session configuration.
def execute_command(self, session, command, options=None)

Execute an HTTP request with OpenID Connect authentication.

Args

session : dict
Session configuration.
command : str
API endpoint path.
options : dict
Optional request parameters (method, params, headers, data, json).

Returns

bytes
Response content.
def open_session(self,
host,
port=None,
login=None,
password=None,
client_id=None,
client_secret=None,
discovery_url=None,
token_endpoint=None,
id_token=None,
access_token=None,
verify_ssl=True)

Opens a session with OpenID Connect authentication.

Args

host : str
Hostname or IP address.
port : int
Port. If None, uses default_port.
login : str
Username for authentication (optional).
password : str
Password for authentication (optional).
client_id : str
OpenID Connect client ID.
client_secret : str
OpenID Connect client secret.
discovery_url : str
OpenID Connect discovery URL (optional).
token_endpoint : str
Token endpoint URL (optional).
id_token : str
Existing ID token (optional).
access_token : str
Existing access token (optional).
verify_ssl : bool
Whether to verify SSL certificates (default: True). Set to False to disable SSL verification for self-signed certificates.

Returns

dict
Session configuration.
class Saml (port=443, use_https=True)

HTTP connector with SAML authentication. Note: SAML authentication typically requires browser-based flow. This implementation supports using SAML assertions/tokens.

Initialize SAML connector.

Args

port : int
Default port (default: 443).
use_https : bool
Whether to use HTTPS (default: True).

Ancestors

Methods

def close_session(self, session)

Close the session (no-op for SAML).

Args

session : dict
Session configuration.
def execute_command(self, session, command, options=None)

Execute an HTTP request with SAML authentication.

Args

session : dict
Session configuration.
command : str
API endpoint path.
options : dict
Optional request parameters (method, params, headers, data, json).

Returns

bytes
Response content.
def open_session(self,
host,
port=None,
login=None,
password=None,
saml_token=None,
saml_header='X-SAML-Token',
verify_ssl=True)

Opens a session with SAML authentication.

Args

host : str
Hostname or IP address.
port : int
Port. If None, uses default_port.
login : str
Not used (for compatibility).
password : str
Not used (for compatibility).
saml_token : str
SAML assertion/token.
saml_header : str
Header name for SAML token (default: X-SAML-Token).
verify_ssl : bool
Whether to verify SSL certificates (default: True). Set to False to disable SSL verification for self-signed certificates.

Returns

dict
Session configuration.