MISP/SkillAegis
MISP/SkillAegis
6
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

new: [backend:misp_api] Added request caching

Browse source
This commit is contained in:
Sami Mokaddem 2024-07-02 13:44:01 +02:00
parent cad62bd3e2
commit af6df877e7
2 changed files with 10 additions and 2 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
REQUIREMENTS
View file

@ -2,4 +2,5 @@ pyzmq
python-socketio python-socketio
eventlet eventlet
requests requests
requests-cache
jq jq

11
misp_api.py
View file

@ -1,15 +1,22 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
import json import json
from datetime import timedelta
from typing import Union from typing import Union
from urllib.parse import urljoin from urllib.parse import urljoin
import requests # type: ignore import requests # type: ignore
import requests.adapters # type: ignore import requests.adapters # type: ignore
from requests_cache import CachedSession
from requests.packages.urllib3.exceptions import InsecureRequestWarning # type: ignore from requests.packages.urllib3.exceptions import InsecureRequestWarning # type: ignore
requests.packages.urllib3.disable_warnings(InsecureRequestWarning) requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
from config import misp_url, misp_apikey, misp_skipssl from config import misp_url, misp_apikey, misp_skipssl
requestSession = CachedSession(cache_name='misp_cache', expire_after=timedelta(seconds=5))
adapterCache = requests.adapters.HTTPAdapter(pool_connections=50, pool_maxsize=50)
requestSession.mount('https://', adapterCache)
requestSession.mount('http://', adapterCache)
def get(url, data={}, api_key=misp_apikey): def get(url, data={}, api_key=misp_apikey):
headers = { headers = {
@ -20,7 +27,7 @@ def get(url, data={}, api_key=misp_apikey):
} }
full_url = urljoin(misp_url, url) full_url = urljoin(misp_url, url)
try: try:
response = requests.get(full_url, data=data, headers=headers, verify=not misp_skipssl) response = requestSession.get(full_url, data=data, headers=headers, verify=not misp_skipssl)
except requests.exceptions.ConnectionError as e: except requests.exceptions.ConnectionError as e:
print('Could not perform request on MISP.', e) print('Could not perform request on MISP.', e)
return None return None
@ -36,7 +43,7 @@ def post(url, data={}, api_key=misp_apikey):
} }
full_url = urljoin(misp_url, url) full_url = urljoin(misp_url, url)
try: try:
response = requests.post(full_url, data=json.dumps(data), headers=headers, verify=not misp_skipssl) response = requestSession.post(full_url, data=json.dumps(data), headers=headers, verify=not misp_skipssl)
except requests.exceptions.ConnectionError as e: except requests.exceptions.ConnectionError as e:
print('Could not perform request on MISP.', e) print('Could not perform request on MISP.', e)
return None return None