Splunk TA UCC Library

https://github.com/splunk/addonfactory-ucc-library
Admin
Splunk TA UCC Library is an open source helper library that extends the Splunk SDK for Python,...
Tokens:5,453
Snippets:34
Trust Score:9.4
License:Apache-2.0
Update:6 months ago
Show doc for...
Tokens:
Raw
### Build demo add-on locally

Source: https://github.com/splunk/addonfactory-ucc-library/blob/main/tests/integration/demo/README.md

Executes the build script to generate an installable archive in the root directory.

```bash
./scripts/build-demo-addon.sh
```

--------------------------------

### AdminExternalHandler for REST Endpoints

Source: https://context7.com/splunk/addonfactory-ucc-library/llms.txt

Use AdminExternalHandler to create custom REST endpoints for Splunk's admin interface. It provides automatic field setup and lifecycle hooks for managing configurations.

```python
from splunktaucclib.rest_handler.admin_external import AdminExternalHandler, handle
from splunktaucclib.rest_handler.endpoint import SingleModel
from splunktaucclib.rest_handler.endpoint.field import RestField
from splunktaucclib.rest_handler.endpoint.validator import String, Number

# Define endpoint fields
fields = [
    RestField(
        name='url',
        required=True,
        validator=String(min_len=1, max_len=500)
    ),
    RestField(
        name='api_key',
        required=True,
        encrypted=True
    ),
    RestField(
        name='interval',
        required=False,
        default='300',
        validator=Number(min_val=60, max_val=86400, is_int=True)
    )
]

# Create the endpoint model
endpoint = SingleModel(
    conf_name='my_addon_account',
    fields=fields,
    user='nobody',
    app='my_addon'
)

# Custom handler with lifecycle hooks
class MyAdminHandler(AdminExternalHandler):

    def create_hook(self, session_key, config_name, stanza_id, payload):
        """Called before creating a new configuration."""
        # Validate or transform payload before creation
        if not payload.get('url', '').startswith('https'):
            payload['url'] = 'https://' + payload.get('url', '')

    def edit_hook(self, session_key, config_name, stanza_id, payload):
        """Called before updating an existing configuration."""
        # Log edit operations
        pass

    def delete_hook(self, session_key, config_name, stanza_id):
        """Called before deleting a configuration."""
        # Cleanup related resources
        pass

# Entry point for the REST endpoint
if __name__ == '__main__':
    handle(endpoint, handler=MyAdminHandler)
```

--------------------------------

### Access Global Settings with Setup_Util

Source: https://context7.com/splunk/addonfactory-ucc-library/llms.txt

Setup_Util provides convenient access to proxy settings, log levels, and credentials, caching them for performance. Initialize with Splunk connection details and a logger. Use methods like get_log_level(), get_proxy_settings(), get_credential_by_id(), and get_customized_setting().

```python
from splunktaucclib.splunk_aoblib.setup_util import Setup_Util
import logging

logger = logging.getLogger('my_addon')

# Initialize setup utility
setup_util = Setup_Util(
    uri='https://localhost:8089',
    session_key='your_session_key',
    logger=logger
)

# Get log level setting
log_level = setup_util.get_log_level()
# Returns: 'INFO', 'DEBUG', 'WARNING', 'ERROR', or 'CRITICAL'

# Get proxy settings (returns empty dict if proxy disabled)
proxy = setup_util.get_proxy_settings()
# {
#     'proxy_url': 'proxy.example.com',
#     'proxy_port': '8080',
#     'proxy_username': 'user',
#     'proxy_password': 'pass',
#     'proxy_type': 'http',
#     'proxy_rdns': True
# }

# Get credential by account ID
credential = setup_util.get_credential_by_id('my_account')
# {'name': 'my_account', 'username': 'admin', 'password': 'secret123'}

# Get credential by username
credential = setup_util.get_credential_by_username('admin')
# {'name': 'account1', 'username': 'admin', 'password': 'secret123'}

# Get custom setting value
custom_value = setup_util.get_customized_setting('my_custom_field')
```

--------------------------------

### Load and Iterate Configurations

Source: https://context7.com/splunk/addonfactory-ucc-library/llms.txt

Loads all configurations and iterates through account data, handling potential configuration errors.

```python
try:
    configurations = config.load()
    # {
    #     '_product': 'my_addon',
    #     '_version': '1.0.0',
    #     'accounts': {
    #         'account1': {'username': 'admin', 'disabled': False, 'timeout': 30}
    #     },
    #     'settings': {...}
    # }

    for account_name, account_data in configurations.get('accounts', {}).items():
        print(f"Account: {account_name}")
        print(f"  Disabled: {account_data.get('disabled')}")  # Boolean
        print(f"  Timeout: {account_data.get('timeout')}")    # Integer

except ConfigException as e:
    print(f"Configuration error: {e}")
```

--------------------------------

### Define and Initialize Configuration Schema

Source: https://context7.com/splunk/addonfactory-ucc-library/llms.txt

Defines the JSON schema for add-on configurations and initializes the Config handler with connection details.

```python
schema = '''
{
    "_product": "my_addon",
    "_rest_namespace": "my_addon",
    "_rest_prefix": "my_addon_",
    "_version": "1.0.0",
    "accounts": {
        "endpoint": "configs/conf-my_addon_account",
        "field_types": {
            "*": {
                "disabled": "bool",
                "timeout": "int",
                "settings": "json"
            }
        }
    },
    "settings": {
        "endpoint": "configs/conf-my_addon_settings"
    }
}
'''

# Initialize config handler
config = Config(
    splunkd_uri='https://localhost:8089',
    session_key='session_key',
    schema=schema,
    user='nobody',
    app='my_addon'
)
```

--------------------------------

### Load and Save Global Configurations with GlobalConfig

Source: https://context7.com/splunk/addonfactory-ucc-library/llms.txt

Use GlobalConfig to manage inputs, accounts, proxy settings, and logging levels. Load configurations using the appropriate methods and save changes by providing a payload. Requires Splunk connection details and a schema.

```python
import json
from splunktaucclib.global_config import GlobalConfig, GlobalConfigSchema

# Load the global configuration schema
with open('appserver/static/js/build/globalConfig.json') as f:
    schema_json = json.load(f)

# Initialize GlobalConfig with Splunk connection details
global_config = GlobalConfig(
    splunkd_uri='https://localhost:8089',
    session_key='your_session_key',
    schema=GlobalConfigSchema(schema_json)
)

# Load all input configurations
inputs = global_config.inputs.load()
# {'my_input_type': [{'name': 'input1', 'interval': '300', ...}]}

# Load specific input type
specific_inputs = global_config.inputs.load(input_type='my_input_type')

# Load account configurations
accounts = global_config.configs.load()
# {'account': [{'name': 'account1', 'username': 'admin', 'password': '***'}]}

# Load global settings (proxy, logging, custom settings)
settings = global_config.settings.load()
# {'settings': [{'name': 'proxy', 'proxy_url': '...', 'proxy_enabled': True}, ...]})

# Save configuration changes
payload = {
    'settings': [
        {'name': 'proxy', 'proxy_host': '1.2.3.4', 'proxy_port': '8080'},
        {'name': 'logging', 'loglevel': 'DEBUG'}
    ]
}
errors = global_config.settings.save(payload)
```

--------------------------------

### Implement a Modular Input with BaseModInput

Source: https://context7.com/splunk/addonfactory-ucc-library/llms.txt

Use BaseModInput to create a modular input that supports configuration parsing, proxy settings, and checkpointing. Ensure the class overrides get_scheme and collect_events to define input behavior.

```python
from splunktaucclib.modinput_wrapper.base_modinput import BaseModInput
from splunklib import modularinput as smi
import sys

class MyModularInput(BaseModInput):
    def __init__(self):
        super().__init__(
            app_namespace='my_addon',
            input_name='my_input',
            use_single_instance=False  # Set True for single instance mode
        )

    def get_app_name(self):
        return 'my_addon'

    def get_scheme(self):
        scheme = super().get_scheme()
        scheme.title = 'My Custom Input'
        scheme.description = 'Collects data from external API'
        scheme.use_external_validation = True

        # Define input arguments
        interval_arg = smi.Argument('interval')
        interval_arg.required_on_create = True
        interval_arg.data_type = smi.Argument.data_type_number
        scheme.add_argument(interval_arg)

        return scheme

    def get_account_fields(self):
        return ['global_account']  # Fields referencing account configs

    def get_checkbox_fields(self):
        return ['ssl_verify']  # Boolean checkbox fields

    def get_global_checkbox_fields(self):
        return ['enable_debug']  # Global boolean settings

    def collect_events(self, event_writer):
        # Get input stanza parameters
        for stanza_name in self.get_input_stanza_names():
            stanza = self.get_input_stanza(stanza_name)

            # Access proxy settings
            proxy = self.get_proxy()

            # Make HTTP request with proxy support
            response = self.send_http_request(
                url='https://api.example.com/data',
                method='GET',
                headers={'Authorization': f"Bearer {stanza.get('api_key')}"},
                verify=stanza.get('ssl_verify', True),
                use_proxy=True
            )

            # Use checkpointing
            last_checkpoint = self.get_check_point(stanza_name)

            # Create and write events
            if response.status_code == 200:
                event = self.new_event(
                    data=response.text,
                    index=self.get_output_index(stanza_name),
                    sourcetype=self.get_sourcetype(stanza_name)
                )
                event_writer.write_event(event)

                # Save checkpoint
                self.save_check_point(stanza_name, {'last_run': 'timestamp'})

if __name__ == '__main__':
    sys.exit(MyModularInput().run(sys.argv))
```

--------------------------------

### Implement CIM Modular Actions in Python

Source: https://context7.com/splunk/addonfactory-ucc-library/llms.txt

Use the ModularAction class to process search results, validate parameters, and generate stash events for Splunk indexing.

```python
from splunktaucclib.cim_actions import ModularAction, ModularActionTimer
import json
import sys
import logging

# Setup logger
logger = ModularAction.setup_logger(
    name='my_modalert',
    level=logging.INFO,
    maxBytes=25000000,
    backupCount=5
)

# Read settings from stdin (provided by Splunk)
settings = sys.stdin.read()

# Initialize modular action
modaction = ModularAction(
    settings=settings,
    logger=logger,
    action_name='my_cim_action'
)

# Load search info
modaction.addinfo()
modaction.addjobinfo()

# Use timer to track execution duration
with ModularActionTimer(modaction, 'main') as timer:
    # Iterate through search results
    for num, result in enumerate(csv.DictReader(open(modaction.results_file, 'r'))):
        # Update modaction with current result identifiers
        result['rid'] = str(num)
        modaction.update(result)

        # Log action invocation
        modaction.invoke()

        # Validate parameters (implement custom validation)
        modaction.validate()

        # Perform action work
        modaction.dowork()

        # Generate stash event for notable indexing
        stash_event = modaction.result2stash(
            result=result,
            addinfo=True  # Include search time bounds
        )

        # Add event with CIM header for tracking
        modaction.addevent(
            raw=stash_event,
            sourcetype='stash_my_action',
            cam_header=True
        )

        # Log success message
        modaction.message(
            signature='Action completed successfully',
            status='success',
            level=logging.INFO
        )

# Write all events to Splunk's stash
success = modaction.writeevents(
    index='summary',
    host=None,  # Auto-detect
    source=None,  # Auto-detect
    fext='my_action_model'
)

# For distributed workers, queue the action instead of executing locally
# modaction.queuework(exit_strategy=True)
```

--------------------------------

### Implement an Alert Action with ModularAlertBase

Source: https://context7.com/splunk/addonfactory-ucc-library/llms.txt

Use ModularAlertBase to build alert actions that require credential access, proxy support, and event processing. The process_event method must be implemented to handle the alert logic.

```python
from splunktaucclib.alert_actions_base import ModularAlertBase
import sys

class MyAlertAction(ModularAlertBase):
    def __init__(self):
        super().__init__(
            ta_name='my_addon',
            alert_name='send_notification'
        )

    def process_event(self):
        # Set log level from global settings
        self.set_log_level(self.log_level)

        # Get alert action parameters
        webhook_url = self.get_param('webhook_url')
        account_id = self.get_param('account')

        # Get credentials
        credential = self.get_user_credential_by_account_id(account_id)
        if not credential:
            self.log_error(f"Account {account_id} not found")
            return 1

        # Get global settings
        custom_setting = self.get_global_setting('custom_field')

        # Get proxy settings
        proxy = self.get_proxy()

        # Process each alert result
        for event in self.get_events():
            # Access event fields
            host = event.get('host', 'unknown')
            message = event.get('message', '')

            # Send HTTP request with proxy support
            try:
                response = self.send_http_request(
                    url=webhook_url,
                    method='POST',
                    payload={'host': host, 'message': message},
                    headers={
                        'Content-Type': 'application/json',
                        'Authorization': f"Bearer {credential['password']}"
                    },
                    use_proxy=True,
                    timeout=30
                )

                if response.status_code == 200:
                    self.log_info(f"Notification sent for {host}")
                else:
                    self.log_error(f"Failed to send: {response.status_code}")

            except Exception as e:
                self.log_error(f"Error sending notification: {str(e)}")
                return 1

        return 0

if __name__ == '__main__':
    alert_action = MyAlertAction()
    sys.exit(alert_action.run(sys.argv))
```

--------------------------------

### UCC Configuration Module

Source: https://context7.com/splunk/addonfactory-ucc-library/llms.txt

The Config class from splunktaucclib.config allows for schema-based configuration loading and saving, including type conversion, for UCC-based add-ons.

```python
from splunktaucclib.config import Config, ConfigException
```

--------------------------------

### Manage Configuration Endpoints with RestHandler

Source: https://context7.com/splunk/addonfactory-ucc-library/llms.txt

Define configuration models with field validation and use RestHandler for CRUD operations on Splunk configuration endpoints.

```python
from splunktaucclib.rest_handler.handler import RestHandler
from splunktaucclib.rest_handler.endpoint import SingleModel, MultipleModel, DataInputModel
from splunktaucclib.rest_handler.endpoint.field import RestField
from splunktaucclib.rest_handler.endpoint.validator import (
    String, Number, Pattern, Host, Port, Email, Enum, RequiresIf
)

# Define a configuration model with validated fields
fields = [
    RestField(
        name='api_url',
        required=True,
        validator=Pattern(r'^https?://.*', flags=0)
    ),
    RestField(
        name='api_key',
        required=True,
        encrypted=True  # Will be stored encrypted
    ),
    RestField(
        name='port',
        required=False,
        default='443',
        validator=Port()
    ),
    RestField(
        name='timeout',
        required=False,
        default='30',
        validator=Number(min_val=1, max_val=300, is_int=True)
    ),
    RestField(
        name='log_level',
        required=False,
        default='INFO',
        validator=Enum(values=['DEBUG', 'INFO', 'WARNING', 'ERROR'])
    ),
    RestField(
        name='email',
        required=False,
        validator=Email()
    ),
    RestField(
        name='proxy_username',
        validator=RequiresIf(
            fields=['proxy_password'],
            condition=lambda value, data: bool(value)
        )
    )
]

# Create endpoint model
endpoint = SingleModel(
    conf_name='my_addon_settings',
    fields=fields,
    user='nobody',
    app='my_addon'
)

# Initialize handler
handler = RestHandler(
    splunkd_uri='https://localhost:8089',
    session_key='session_key',
    endpoint=endpoint
)

# Create new configuration
entity = list(handler.create('my_config', {
    'api_url': 'https://api.example.com',
    'api_key': 'secret_key_123',
    'port': '443',
    'timeout': '60'
}))[0]

# Get configuration (credentials masked by default)
entity = list(handler.get('my_config'))[0]
print(entity.content)
# {'api_url': 'https://...', 'api_key': '******', 'port': '443', ...}

# Get with decrypted credentials
entity = list(handler.get('my_config', decrypt=True))[0]
print(entity.content['api_key'])  # 'secret_key_123'

# List all configurations
for entity in handler.all():
    print(f"{entity.name}: {entity.content}")

# Update configuration
list(handler.update('my_config', {'timeout': '120'}))

# Enable/disable configuration
list(handler.disable('my_config'))
list(handler.enable('my_config'))

# Delete configuration
list(handler.delete('my_config'))
```

--------------------------------

### Update Configuration Items

Source: https://context7.com/splunk/addonfactory-ucc-library/llms.txt

Updates specific fields for multiple configuration items, with an option to suppress exceptions on failure.

```python
failed_items = config.update_items(
    endpoint_id='accounts',
    item_names=['account1', 'account2'],
    field_names=['timeout', 'disabled'],
    data={
        'account1': {'timeout': 60, 'disabled': False},
        'account2': {'timeout': 120, 'disabled': True}
    },
    raise_if_failed=False
)

if failed_items:
    print(f"Failed to update: {failed_items}")
```

--------------------------------

### Field Validators for REST Endpoints

Source: https://context7.com/splunk/addonfactory-ucc-library/llms.txt

Utilize a comprehensive set of validators for input validation in Splunk REST handlers and configuration endpoints. Supports string, number, pattern, host, port, email, datetime, JSON, enum, and composite validations.

```python
from splunktaucclib.rest_handler.endpoint.validator import (
    Validator, String, Number, Pattern, Host, Port, Email,
    Datetime, JsonString, Enum, AnyOf, AllOf, RequiresIf,
    UserDefined,
    ValidationFailed
)

# String length validation
username_validator = String(min_len=3, max_len=50)

# Number range validation
port_validator = Number(min_val=1, max_val=65535, is_int=True)
timeout_validator = Number(min_val=0.1, max_val=300.0, is_int=False)

# Pattern matching
url_validator = Pattern(r'^https?://[\w\-.]+(:\d+)?(/.*)?$')
ip_validator = Pattern(r'^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$')

# Built-in validators
host_validator = Host()  # Validates hostname format
port_validator = Port()  # Validates 0-65535
email_validator = Email()  # Validates email format

# Datetime format validation
date_validator = Datetime(datetime_format='%Y-%m-%d')
timestamp_validator = Datetime(datetime_format='%Y-%m-%dT%H:%M:%S.%f')

# JSON string validation
json_validator = JsonString()

# Enumeration validation
log_level_validator = Enum(values=['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'])
auth_type_validator = Enum(values=['basic', 'oauth2', 'api_key'])

# Composite validators
# AnyOf - passes if any validator passes
flexible_port = AnyOf(
    Port(),
    Enum(values=['default', 'auto'])
)

# AllOf - passes only if all validators pass
strict_password = AllOf(
    String(min_len=8, max_len=128),
    Pattern(r'.*[A-Z].*'),  # At least one uppercase
    Pattern(r'.*[0-9].*')   # At least one digit
)

# Conditional requirements
proxy_password_validator = RequiresIf(
    fields=['proxy_password'],  # These fields are required
    condition=lambda value, data: bool(data.get('proxy_username'))
)

# Custom validator function
def validate_api_endpoint(value, data, *args, **kwargs):
    """Custom validation logic."""
    if not value.startswith('https://'):
        raise ValidationFailed('API endpoint must use HTTPS')
    if 'api' not in value:
        raise ValidationFailed('Invalid API endpoint format')

custom_validator = UserDefined(validate_api_endpoint)

# Using validators in RestField
from splunktaucclib.rest_handler.endpoint.field import RestField

field = RestField(
    name='password',
    required=True,
    encrypted=True,
    validator=AllOf(
        String(min_len=8),
        Pattern(r'.*[!@#$%^&*].*')  # Require special character
    )
)
```
Tokens:
Raw