Search & Query

def search(
    self,
    controller: str = 'events',
    return_format: str = 'json',
    limit: int = None,
    page: int = None,
    value: Union[str, List[str]] = None,
    type_attribute: Union[str, List[str]] = None,
    category: Union[str, List[str]] = None,
    org: Union[str, int, List] = None,
    tags: Union[str, List[str]] = None,
    not_tags: Union[str, List[str]] = None,
    date_from: Union[str, datetime] = None,
    date_to: Union[str, datetime] = None,
    last: str = None,
    eventid: Union[str, int, List] = None,
    with_attachments: bool = None,
    metadata: bool = None,
    uuid: Union[str, List[str]] = None,
    publish_timestamp: Union[str, int] = None,
    timestamp: Union[str, int, List] = None,
    enforceWarninglist: bool = None,
    to_ids: bool = None,
    deleted: bool = False,
    include_event_uuid: bool = None,
    include_event_tags: bool = None,
    event_timestamp: Union[str, int] = None,
    sg_reference_only: bool = None,
    eventinfo: str = None,
    searchall: str = None,
    requested_attributes: List[str] = None,
    include_context: bool = None,
    headerless: bool = None,
    include_sightings: bool = None,
    include_correlations: bool = None,
    include_decay_score: bool = None,
    decayingModel: Union[str, int, List] = None,
    exclude_decayed: bool = None,
    score: int = None,
    **kwargs
) -> Union[List, Dict]:
    """
    Perform comprehensive search across MISP data.
    
    Parameters:
    - controller: Search target ('events', 'attributes', 'objects')
    - return_format: Output format ('json', 'xml', 'csv', 'stix', 'stix2', 'yara', etc.)
    - limit: Maximum results to return
    - page: Page number for pagination
    - value: Search by attribute values (supports wildcards)
    - type_attribute: Filter by attribute types
    - category: Filter by attribute categories
    - org: Filter by organization ID/name
    - tags: Include only items with these tags
    - not_tags: Exclude items with these tags
    - date_from: Start date for search range
    - date_to: End date for search range
    - last: Time range (e.g., '5d', '2h', '30m')
    - eventid: Filter by specific event IDs
    - with_attachments: Include only items with attachments
    - metadata: Return metadata only (faster)
    - uuid: Filter by specific UUIDs
    - publish_timestamp: Filter by publish time
    - timestamp: Filter by modification time
    - enforceWarninglist: Apply warning list filtering
    - to_ids: Filter by IDS export flag
    - deleted: Include deleted items
    - include_event_uuid: Include event UUID in results
    - include_event_tags: Include event tags in results
    - event_timestamp: Filter by event timestamp
    - sg_reference_only: Sharing group reference only
    - eventinfo: Search in event info field
    - searchall: Search across all fields
    - requested_attributes: Specific attributes to return
    - include_context: Include contextual information
    - headerless: Return results without headers
    - include_sightings: Include sighting data
    - include_correlations: Include correlation data
    - include_decay_score: Include decay scoring
    - decayingModel: Specific decaying models to use
    - exclude_decayed: Exclude decayed indicators
    - score: Minimum decay score threshold
    
    Returns:
    Search results in specified format
    """

def search_index(
    self,
    all: str = None,
    attribute: str = None,
    email: str = None,
    published: bool = None,
    hasproposal: bool = None,
    eventid: Union[str, int] = None,
    datefrom: str = None,
    dateto: str = None,
    organization: str = None,
    tag: str = None,
    **kwargs
) -> List[Dict]:
    """
    Search event index with text-based filtering.
    
    Parameters:
    - all: Search across all fields
    - attribute: Search in attributes
    - email: Search by email addresses
    - published: Filter by published status
    - hasproposal: Filter events with proposals
    - eventid: Specific event ID
    - datefrom: Start date
    - dateto: End date
    - organization: Organization filter
    - tag: Tag filter
    
    Returns:
    List of matching events (index format)
    """

def search_sightings(
    self,
    context: str = None,
    context_id: Union[str, int] = None,
    type_sighting: str = None,
    date_from: Union[str, datetime] = None,
    date_to: Union[str, datetime] = None,
    last: str = None,
    org: Union[str, int] = None,
    source: str = None,
    include_attribute: bool = None,
    include_event: bool = None,
    **kwargs
) -> List[Dict]:
    """
    Search sighting data.
    
    Parameters:
    - context: Sighting context ('attribute', 'event')
    - context_id: ID of context object
    - type_sighting: Sighting type (0=sighting, 1=false positive, 2=expiration)
    - date_from: Start date for sightings
    - date_to: End date for sightings
    - last: Time range
    - org: Organization filter
    - source: Sighting source
    - include_attribute: Include attribute data
    - include_event: Include event data
    
    Returns:
    List of sighting records
    """

def search_logs(
    self,
    limit: int = None,
    page: int = None,
    log_id: Union[int, str] = None,
    title: str = None,
    created: str = None,
    model: str = None,
    action: str = None,
    user_id: Union[int, str] = None,
    change: str = None,
    email: str = None,
    org: str = None,
    description: str = None,
    ip: str = None,
    **kwargs
) -> List[Dict]:
    """
    Search system logs.
    
    Parameters:
    - limit: Maximum logs to return
    - page: Page number
    - log_id: Specific log ID
    - title: Log title filter
    - created: Creation date filter
    - model: MISP model filter
    - action: Action type filter
    - user_id: User ID filter
    - change: Change description filter
    - email: User email filter
    - org: Organization filter
    - description: Description filter
    - ip: IP address filter
    
    Returns:
    List of log entries
    """

def search_tags(
    self,
    tagname: str = None,
    **kwargs
) -> List[Dict]:
    """
    Search available tags.
    
    Parameters:
    - tagname: Tag name filter (supports wildcards)
    
    Returns:
    List of matching tags
    """

def search_galaxy(
    self,
    galaxy_name: str = None,
    **kwargs
) -> List[Dict]:
    """
    Search galaxy knowledge base.
    
    Parameters:
    - galaxy_name: Galaxy name filter
    
    Returns:
    List of matching galaxies
    """

def search_galaxy_clusters(
    self,
    galaxy_id: Union[int, str] = None,
    cluster_value: str = None,
    **kwargs
) -> List[Dict]:
    """
    Search galaxy clusters.
    
    Parameters:
    - galaxy_id: Specific galaxy ID
    - cluster_value: Cluster value filter
    
    Returns:
    List of matching galaxy clusters
    """

def build_complex_query(
    self,
    or_parameters: List[Dict] = None,
    and_parameters: List[Dict] = None,
    not_parameters: List[Dict] = None
) -> Dict:
    """
    Build complex search query with logical operators.
    
    Parameters:
    - or_parameters: OR conditions
    - and_parameters: AND conditions  
    - not_parameters: NOT conditions
    
    Returns:
    Complex query structure
    """

def get_csv(
    self,
    eventid: Union[str, int, List] = None,
    ignore: bool = None,
    tags: Union[str, List[str]] = None,
    category: Union[str, List[str]] = None,
    type_attribute: Union[str, List[str]] = None,
    include_context: bool = True,
    **kwargs
) -> str:
    """
    Export search results as CSV.
    
    Parameters:
    - eventid: Event IDs to export
    - ignore: Ignore warning list
    - tags: Tag filters
    - category: Category filters
    - type_attribute: Type filters
    - include_context: Include contextual data
    
    Returns:
    CSV formatted string
    """

from pymisp import PyMISP
from datetime import datetime, timedelta

misp = PyMISP('https://misp.example.com', 'your-api-key')

# Search for specific IP address
ip_results = misp.search(value='192.168.1.100')

# Search for domain patterns
domain_results = misp.search(
    value='%.malware.com',
    type_attribute='domain'
)

# Search recent events
recent_events = misp.search(
    controller='events',
    last='7d',
    published=True
)

# Search with multiple filters
advanced_results = misp.search(
    controller='attributes',
    type_attribute=['ip-dst', 'domain', 'url'],
    category='Network activity',
    tags=['apt', 'malware'],
    not_tags=['false-positive'],
    org=[1, 2, 3],
    to_ids=True,
    last='30d'
)

# Search with date range
from datetime import datetime, timedelta

end_date = datetime.now()
start_date = end_date - timedelta(days=30)

date_filtered = misp.search(
    date_from=start_date,
    date_to=end_date,
    published=True
)

# Search across all fields
text_search = misp.search(searchall='APT campaign')

# Search event info
info_search = misp.search(
    controller='events',
    eventinfo='*malware*'
)

# Search with index
index_results = misp.search_index(
    all='targeted attack',
    published=True,
    datefrom='2024-01-01'
)

# Export as different formats
stix_export = misp.search(
    eventid=[1, 2, 3],
    return_format='stix2'
)

csv_export = misp.search(
    tags=['apt'],
    return_format='csv',
    include_context=True
)

# Get CSV directly
csv_data = misp.get_csv(
    eventid=[1, 2, 3],
    category=['Network activity', 'Payload delivery'],
    include_context=True
)

# Search sightings
sightings = misp.search_sightings(
    context='attribute',
    last='7d',
    include_attribute=True
)

# Search system logs
logs = misp.search_logs(
    action='add',
    model='Event',
    user_id=1,
    last='24h'
)

# Search tags
matching_tags = misp.search_tags(tagname='apt*')

# Build complex OR query
complex_query = misp.build_complex_query(
    or_parameters=[
        {'type': 'ip-dst', 'value': '192.168.1.100'},
        {'type': 'domain', 'value': 'malware.com'},
        {'type': 'url', 'value': '*evil.com*'}
    ],
    and_parameters=[
        {'tags': 'apt'},
        {'published': True}
    ],
    not_parameters=[
        {'tags': 'false-positive'}
    ]
)

results = misp.search(**complex_query)

# Search with correlations
correlated_search = misp.search(
    value='192.168.1.100',
    include_correlations=True,
    include_sightings=True
)

# Search with decay scoring
decay_search = misp.search(
    tags=['apt'],
    include_decay_score=True,
    score=50,  # Minimum score
    exclude_decayed=True
)

# Context-aware search
context_search = misp.search(
    value='malware.exe',
    include_context=True,
    include_event_tags=True,
    requested_attributes=['md5', 'sha1', 'filename']
)

from typing import Union, List, Dict, Optional, Any
from datetime import datetime

SearchValue = Union[str, int, List[Union[str, int]]]
SearchFilter = Dict[str, Any]
SearchResults = Union[List[Dict], Dict]
ExportFormat = str  # 'json', 'xml', 'csv', 'stix', 'stix2', 'yara', etc.
TimeRange = str  # '5d', '2h', '30m', etc.