Platform Integrations API Code Examples

This page provides practical Python examples to help you get started with the NordStellar Platform Integrations API. These examples show common API operations to help you integrate NordStellar's security data into your tools and workflows.

Authentication

All API requests require authentication using a Bearer token. Make sure to replace YOUR_API_KEY in the examples with your actual API key generated from the Settings page.

Example 1: Fetching All Project IDs

Before using other API endpoints, you'll need to know your project IDs. This example shows how to retrieve all projects associated with your account:

import requests
 
# Configuration
API_KEY = "YOUR_API_KEY"
BASE_URL = "https://platform-integration-api.nordstellar.com"
 
# Set up headers with authentication
headers = {
    "Authorization": f"Bearer {API_KEY}",
    "Content-Type": "application/json"
}
 
# Define request parameters (optional)
params = {
    "limit": 100  # Adjust as needed
}
 
# Make API request
response = requests.get(
    f"{BASE_URL}/v1/projects",
    headers=headers,
    params=params
)
 
# Process response
if response.status_code == 200:
    data = response.json()
    projects = data['items']
    print(f"Retrieved {len(projects)} projects:")
    
    for project in projects:
        print(f"Project ID: {project['id']}")
        print(f"Project Name: {project['name']}")
        print("-" * 50)
    
    # Handle pagination if needed
    if data.get('next'):
        print(f"More projects available at: {data['next']}")
else:
    print(f"Error: {response.status_code}")
    print(response.text)

Example 2: Retrieving Project Events

Once you have your project ID, you can retrieve the security events associated with that project:

import requests
import datetime
 
# Configuration
API_KEY = "YOUR_API_KEY"
BASE_URL = "https://platform-integration-api.nordstellar.com"
PROJECT_ID = "52c6fd64-8e91-46f2-ba5a-8761e103f9cf"  # Replace with your project ID
 
# Set up headers with authentication
headers = {
    "Authorization": f"Bearer {API_KEY}",
    "Content-Type": "application/json"
}
 
# Define date range (last 30 days)
end_date = datetime.datetime.now().isoformat() + "Z"
start_date = (datetime.datetime.now() - datetime.timedelta(days=30)).isoformat() + "Z"
 
# Define request parameters
params = {
    "date-added-from": start_date,
    "date-added-to": end_date,
    "limit": 100
}
 
# Make API request
response = requests.get(
    f"{BASE_URL}/v1/projects/{PROJECT_ID}/events",
    headers=headers,
    params=params
)
 
# Process response
if response.status_code == 200:
    events = response.json()
    print(f"Retrieved {len(events['items'])} events")
    
    for event in events['items']:
        print(f"ID: {event['id']}")
        print(f"Type: {event['type']}")
        print(f"Risk Level: {event['risk_level']}")
        print(f"Date Added: {event['date_added']}")
        print("-" * 50)
        
    # Handle pagination if needed
    if events.get('next'):
        print(f"More events available at: {events['next']}")
else:
    print(f"Error: {response.status_code}")
    print(response.text)

Example 3: Resolving an Event

You can mark events as resolved using the following code:

import requests
 
# Configuration
API_KEY = "YOUR_API_KEY"
BASE_URL = "https://platform-integration-api.nordstellar.com"
EVENT_ID = "55cfd1a8-9285-4fd9-bdfe-15a217d77d1f"  # Replace with your event ID
 
# Set up headers with authentication
headers = {
    "Authorization": f"Bearer {API_KEY}",
    "Content-Type": "application/json"
}
 
# Define request body
payload = {
    "is_resolved": True
}
 
# Make API request
response = requests.patch(
    f"{BASE_URL}/v1/events/{EVENT_ID}/is_resolved",
    headers=headers,
    json=payload
)
 
# Process response
if response.status_code == 200:
    result = response.json()
    print(f"Event {result['id']} resolved status: {result['is_resolved']}")
else:
    print(f"Error: {response.status_code}")
    print(response.text)

Example 4: Getting Detailed Information About a Malware Infection Event

This example shows how to retrieve detailed information about a specific malware infection event:

import requests
import json
 
# Configuration
API_KEY = "YOUR_API_KEY"
BASE_URL = "https://platform-integration-api.nordstellar.com"
EVENT_ID = "52c6fd64-8e91-46f2-ba5a-8761e103f9cf"  # Replace with your malware infection event ID
 
# Set up headers with authentication
headers = {
    "Authorization": f"Bearer {API_KEY}",
    "Content-Type": "application/json"
}
 
# Make API request
response = requests.get(
    f"{BASE_URL}/v1/events/malware-infections/{EVENT_ID}",
    headers=headers
)
 
# Process response
if response.status_code == 200:
    event = response.json()
    print(f"Malware Infection Event Details:")
    print(f"ID: {event['id']}")
    print(f"Stealer Name: {event['stealer_name']}")
    print(f"Risk Level: {event['risk_level']}")
    print(f"Date Added: {event['date_added']}")
    print(f"Infection Date: {event['infection_date']}")
    print(f"Asset: {event['asset']['type']} - {event['asset']['value']}")
    
    # Print credentials if available
    if 'credentials' in event and event['credentials']:
        print("\nCompromised Credentials:")
        for cred in event['credentials']:
            if cred['is_corporate']:
                print(f"URL: {cred['url']}")
                print(f"Email: {cred['email']}")
                print(f"Username: {cred['username']}")
                print(f"Application: {cred['application']}")
                print("-" * 40)
else:
    print(f"Error: {response.status_code}")
    print(response.text)

Example 5: SIEM Integration (Splunk)

This example shows how to integrate NordStellar events with Splunk using the HTTP Event Collector (HEC):

import requests
import datetime
import time
 
# NordStellar Configuration
NORDSTELLAR_API_KEY = "YOUR_NORDSTELLAR_API_KEY"
NORDSTELLAR_BASE_URL = "https://platform-integration-api.nordstellar.com"
PROJECT_ID = "52c6fd64-8e91-46f2-ba5a-8761e103f9cf"  # Replace with your project ID
 
# Splunk Configuration
SPLUNK_HEC_URL = "https://your-splunk-instance:8088/services/collector"
SPLUNK_HEC_TOKEN = "YOUR_SPLUNK_HEC_TOKEN"
SPLUNK_INDEX = "nordstellar"  # Create this index in Splunk
 
# Set up NordStellar API headers
nordstellar_headers = {
    "Authorization": f"Bearer {NORDSTELLAR_API_KEY}",
    "Content-Type": "application/json"
}
 
# Set up Splunk HEC headers
splunk_headers = {
    "Authorization": f"Splunk {SPLUNK_HEC_TOKEN}",
    "Content-Type": "application/json"
}
 
def fetch_and_forward_events():
    # Calculate time range (last hour)
    end_date = datetime.datetime.now().isoformat() + "Z"
    start_date = (datetime.datetime.now() - datetime.timedelta(hours=1)).isoformat() + "Z"
    
    # Define request parameters
    params = {
        "date-added-from": start_date,
        "date-added-to": end_date,
        "limit": 100
    }
    
    # Make API request to NordStellar
    response = requests.get(
        f"{NORDSTELLAR_BASE_URL}/v1/projects/{PROJECT_ID}/events",
        headers=nordstellar_headers,
        params=params
    )
    
    if response.status_code != 200:
        print(f"Error fetching events: {response.status_code}")
        print(response.text)
        return
    
    events = response.json()
    print(f"Retrieved {len(events['items'])} events")
    
    # Forward events to Splunk
    for event in events['items']:
        # Prepare event for Splunk
        splunk_event = {
            "time": int(datetime.datetime.fromisoformat(event['date_added'].replace('Z', '')).timestamp()),
            "source": "nordstellar",
            "sourcetype": f"nordstellar:{event['type'].lower()}",
            "index": SPLUNK_INDEX,
            "event": event
        }
        
        # Send to Splunk HEC
        splunk_response = requests.post(
            SPLUNK_HEC_URL,
            headers=splunk_headers,
            json=splunk_event,
            verify=False  # Note: In production, use proper certificate verification
        )
        
        if splunk_response.status_code != 200:
            print(f"Error sending to Splunk: {splunk_response.status_code}")
            print(splunk_response.text)
        else:
            print(f"Successfully sent event {event['id']} to Splunk")
 
if __name__ == "__main__":
    # Run once on script execution
    fetch_and_forward_events()
    
    # Uncomment the following to run as a scheduled process
    """
    while True:
        fetch_and_forward_events()
        print("Sleeping for 1 hour...")
        time.sleep(3600)  # Run every hour
    """

Additional Resources

For more information about the available endpoints and response formats, refer to our Swagger documentation.

API Reference Getting Started