Hello everyone!
I would like to to create detections with unicode characters in the description but I cant seem to find a way to make the API work with unicode characters.
Anyone had the same problem?
Thanks in advance!
Hi @gbL2k, thanks for reaching out!
Could you post here an example of the request that you are making (endpoint, payload, etc), and the error that you are getting back?
I can ,sure.
The endpoint is /api/detection_engine/rules with a POST.
Here is an example payload:
{
"author": ["test"],
"description": "test",
"from": "now-6m",
"name": "test íú",
"risk_score": 100,
"severity": "medium",
"index": ["logs-*"],
"type": "query",
"rule_id": "00000000-0000-0000-0000-00000000000f",
"language": "lucene",
"query": "(process.executable:*) ",
"false_positives": ["test"],
"note": "test: áéő ",
"threat": [{"framework": "MITRE ATT&CK", "tactic": {"id": "TA0006", "name": "Credential Access", "reference": "https://attack.mitre.org/tactics/TA0006/"}, "technique": [{"id": "T1003", "name": "OS Credential Dumping", "reference": "https://attack.mitre.org/techniques/T1003/", "subtechnique": [{"id": "T1003.001", "name": "OS Credential Dumping: LSASS Memory", "reference": "https://attack.mitre.org/techniques/T1003/001/"}]}]}],
"enabled": true
}
It only fails if I have Unicode characters in the note or description field.
The response is 400.
Hi @gbL2k . I just ran that request to create a rule with your exact same payload and the rule was created succesfully:
I think we need to evaluate if the 400 response is being cause by some other issue in the body payload or headers.
A couple questions:
- do you get an error message along with the 400 status code? How are you running this request? curl, Postman, a script, etc?
- what does your Elastic setup look like? Are you self-managed or Cloud?
- How are you authenticating your request?
Thanks for the answer it helps a lot,now I know I have to look somewhere else.
We have tested and its working from postman.
However if we want to make the request from python it will not work.
The elastic setup is self-managed.
We use API key authentication.
The request in python looks as the following:
elastic_data=requests.post(url,headers=headers, data=data,verify=False)
where the data is the one seen above and the headers are the following:
headers = {
'Content-Type': 'application/json;charset=UTF-8',
'kbn-xsrf': 'true',
'Authorization': 'ApiKey ' + api_key
}
So I can help you better, would you mind pasting the whole code snippet of the request? Of course, redact the API key.
Also, what error messages do you see? Can you add some logging?
Sure,its quite simple and short, most of the scripts is only building a valid json object.Which it does perfectly.Unfortunately I only see a 400 response and nothing else.
import requests
import os
import tomllib
import sys
import json
#example usage: update_alert.py Detections https://127.0.0.1:5601/api/detection_engine/rules API_KEY
url = sys.argv[2]
api_key = sys.argv[3]
headers = {
'Content-Type': 'application/json;charset=UTF-8',
'kbn-xsrf': 'true',
'Authorization': 'ApiKey ' + api_key
}
directory=sys.argv[1]
files = os.listdir(directory)
for file_name in files:
file_path=os.path.join(directory,file_name)
if file_path.endswith(".toml"):
with open(file_path,"rb") as toml:
alert=tomllib.load(toml)
data=""
data+="{\n"
for field in alert['rule']:
if type(alert['rule'][field]) == list:
data += " " + "\"" + field + "\": " + str(alert['rule'][field]).replace("'","\"") + "," + "\n"
elif type(alert['rule'][field]) == str:
if field == 'description':
data += " " + "\"" + field + "\": \"" + str(alert['rule'][field]).replace("\n"," ").replace("\"","\\\"").replace("\\","\\\\") + "\"," + "\n"
elif field == 'query':
data += " " + "\"" + field + "\": \"" + str(alert['rule'][field]).replace("\\","\\\\").replace("\"","\\\"").replace("\n"," ") + "\"," + "\n"
else:
data += " " + "\"" + field + "\": \"" + str(alert['rule'][field]).replace("\n"," ").replace("\"","\\\"") + "\"," + "\n"
elif type(alert['rule'][field]) == int:
data += " " + "\"" + field + "\": " + str(alert['rule'][field]) + "," + "\n"
elif type(alert['rule'][field]) == dict:
data += " " + "\"" + field + "\": " + str(alert['rule'][field]).replace("'","\"") + "," + "\n"
data+=" \"enabled\": true\n}\n"
print(data)
elastic_data=requests.post(url,headers=headers, data=data,verify=False)
print(elastic_data)
Can you post here the results from
print(data)
print(elastic_data)
?
The output of data is the json seen above and the output of elastic_data is actually the response from elastic which is "<Response [400]>"
Maybe the problem is actually in python and the requests library?
Maybe the problem is actually in python and the requests library?
It's possible. The payload looks correctly formatted.
Can you try adding the following header?
elastic-api-version: 2023-10-31
Also, what about the URL? Can you print that out? Just to do a sanity check that we are pointing to the expected endpoint.
I have added the header and nothing changed.
The uri is the following:
https://ELASTIC_IP:5601/api/detection_engine/rules
Try:
https://ELASTIC_IP:5601/kbn/api/detection_engine/rules
Does the response change to 404?
No, its remained the same.
Ok, let's try to get more information from the response. Try adding:
try:
error_message = elastic_data.json()
print("Error Message:", error_message)
except json.JSONDecodeError:
print("Error decoding JSON in response body.")
I did and got the answer:""Error decoding JSON in response body."
We have also tried to gather some more information why the request failed with this exact method but it seems we only get a response code back.
Ok, so I created a Python script and replicated the issue.
It's a simpler issue 
: you need to properly format you data as JSON before passing it to the requests library:
import json
# rest of code...
elastic_data = requests.post(url, headers=headers, data=json.dumps(data), verify=False)
I got 200 with this fix, and a properly created rule, with Unicode characters.
Also, for next time when debugging the response, use print(elastic_data.__dict__)
Then I will offically go insanse over why its still not working.
We have also tried this method before.
Would you mind pasting the code here?
Either way thanks for the help Pablo.
This works for me and I think should work for you as well:
import requests
import json
headers = {
'Content-Type': 'application/json;charset=UTF-8',
'kbn-xsrf': 'true',
'Authorization': "", # Your auth here
'elastic-api-version': '2023-10-31'
}
data = {
"author": ["test"],
"description": "test",
"from": "now-6m",
"name": "test íú",
"risk_score": 100,
"severity": "medium",
"index": ["logs-*"],
"type": "query",
"rule_id": "00000000-0000-0000-0000-0000adfa0000000f",
"language": "lucene",
"query": "(process.executable:*) ",
"false_positives": ["test"],
"note": "test: áéő ",
"threat": [
{
"framework": "MITRE ATT&CK",
"tactic": {
"id": "TA0006",
"name": "Credential Access",
"reference": "https://attack.mitre.org/tactics/TA0006/"
},
"technique": [
{
"id": "T1003",
"name": "OS Credential Dumping",
"reference": "https://attack.mitre.org/techniques/T1003/",
"subtechnique": [
{
"id": "T1003.001",
"name": "OS Credential Dumping: LSASS Memory",
"reference": "https://attack.mitre.org/techniques/T1003/001/"
}
]
}
]
}
],
"enabled": True
}
# I'm running Kibana locally, but use your own url
url = 'http://localhost:5601/kbn/api/detection_engine/rules'
elastic_data = requests.post(url, headers=headers, data=json.dumps(data), verify=False)
print(elastic_data.__dict__)