| | import requests |
| | from bs4 import BeautifulSoup |
| | from elasticsearch import Elasticsearch |
| | from langchain.tools import tool |
| |
|
| | es = Elasticsearch( |
| | "https://localhost:9200", |
| | basic_auth=("elastic","dVJI85*y60R3ZVbECj1w"), |
| | ca_certs="/Volumes/macOS/Projects/PFE UM6P/elasticsearch-8.12.1/config/certs/http_ca.crt" |
| | ) |
| |
|
| |
|
| | class EventSearchTool(): |
| | @tool("Event search Tool") |
| | def search(keyword: str): |
| | """Useful tool to search for an indicator of compromise or an security event |
| | Parameters: |
| | - keyword: The keyword to search for |
| | Returns: |
| | - A list of events that match the keyword |
| | """ |
| |
|
| | |
| | |
| | |
| | |
| | query = { |
| | "match": {"value": { |
| | "query": keyword |
| | }} |
| | } |
| |
|
| | |
| | res = es.search(size=5, index="all_events_full", query=query, knn=None, _source=["event_id", "event_title", "event_date", "category", "attribute_tags", "type", "value"]) |
| | hits = res["hits"]["hits"] |
| | events = [x['_source'] for x in hits] |
| |
|
| | return events |
| | |
| |
|
| | @tool("Event search by event_id Tool") |
| | def get_event_by_id(id:str): |
| | """Useful tool to search for an event by its id, and return the full event details |
| | Parameters: |
| | - id: The event id to search for |
| | Returns: |
| | - The full details of the event with the specified id |
| | """ |
| |
|
| | if not es.ping(): |
| | raise "ElasticNotReachable" |
| | res = es.search(index="all_events_full", query={"match": {"event_id": id}}, _source=["event_id", "event_title", "event_date", "category", "attribute_tags", "type", "value"]) |
| | hits = res["hits"]["hits"] |
| | events = [x['_source'] for x in hits] |
| |
|
| | return events |
| | |
| |
|
