I’m working on a custom Splunk app (VendorThreatIntel) that ingests alert data from an external API using the Splunk Python SDK. Before inserting events into a custom index, I perform a duplicate check based on a unique id field. For each incoming record, I run a one-shot search against the index to check whether the record already exists. If found, I skip insertion; otherwise, I insert the event. Below is the logic I’m using: if 'data' in response and isinstance(response['data'], Sequence): inserted_count = 0 skipped_count = 0 logger.info("[Vendor Events] Received %s records for processing", len(response['data'])) server = client.connect(token=session_key, app="VendorThreatIntel") for row in response['data']: row['service_name'] = service['displayName'] row_id = row.get("id") search_query = f'search index="vendor_alerts" id="{row_id}"' logger.info(f"[Vendor Events] Checking for existing record: {search_query}") results_found = False try: rr = results.JSONResultsReader( server.jobs.oneshot(search_query, output_mode="json") ) for result in rr: if isinstance(result, dict): logger.info( "[Vendor Events] Duplicate record found with id=%s, skipping insertion.", row_id ) results_found = True skipped_count += 1 if not results_found: index = server.indexes["vendor_alerts"] index.submit(json.dumps(row)) inserted_count += 1 except Exception as e: logger.error("[Vendor Events] Error inserting data: %s", str(e)) logger.info( "[Vendor Events] Summary | Total: %s | Inserted: %s | Skipped: %s", len(response['data']), inserted_count, skipped_count ) Even when records with the same id already exist in the index, the duplicate detection logic is not being applied, and duplicate events are still getting indexed. May I know why this duplication logic is not being applied? Is this related to one-shot search behavior, indexing latency, or event availability timing during ingestion? Additionally, I would like guidance considering Splunk’s distributed architecture (search head and indexers). Is there any other recommended or reliable way to handle duplicate detection in Splunk, especially in a distributed environment? Thanks in advance. ... View more

We are storing data in a Splunk lookup file on one of the forwarders.  In our distributed Splunk architecture, this lookup data is not getting forwarded to the indexers or the search head, and therefore it is not available for search or enrichment.  How can we sync or transfer this lookup data from the forwarder to the search head (or indexers) so that it can be used across the distributed environment?   ... View more

url = "https://xyz.com/core/api-ua/user-account/stix/v2.1?isSafe=false&key=key" # Path to your custom CA bundle (optional, if you need to use a specific CA bundle) ca_bundle_path = "/home/ubuntu/splunk/etc/apps/APP-Name/certs/ap.pem" # Make the API call through the HTTPS proxy with SSL verification response = requests.get(url, proxies=proxies, verify=ca_bundle_path) print("Response content:", response.content) If I use this code in separate python script.. It works and gives the response. However, If I use the same code in splunk, It doesn't. I get : SSLError(SSLCertVerificationError(1, '[SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: unable to get issuer certificate (_ssl.c:1106)')))   The code that is being used is : files = os.path.join(os.environ['SPLUNK_HOME'], 'etc', 'apps', 'App-Name', 'certs') pem_files = [f"{files}/{file}" for file in os.listdir(path=files) if (file.endswith('.pem') or file.endswith('.crt'))] url = f"{url}/core/api-ua/v2/alerts/attack-surface?type=open-ports&size=1&key={api_token}" if pem_files: logger.info(f"Certificate used: {pem_files[0]}") logger.info(requests.__version__) logger.info(urllib3.__version__) logger.info(proxy_settings) response = requests.request( GET, url, verify=pem_files[0], proxies=proxy_settings ) response.raise_for_status()   In the place of verify=pem_files[0],I have added verify="/home/ubuntu/splunk/etc/apps/APP-Name/certs/ap.pem" Still same error. ... View more