tannedcung/sbt-idp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add: semi auto to api caller

Browse Source
This commit is contained in:
TannedCung 2024-07-17 17:45:12 +07:00
parent f609c53267
commit 0205071f46
2 changed files with 36 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
api-cronjob/Dockerfile
View File

@ -3,8 +3,10 @@ FROM python:3.9-slim
WORKDIR /app
COPY run.py .
COPY requirements.txt .
RUN apt-get update && apt-get -y install curl
RUN pip install -r requirements.txt
RUN pip install requests
CMD [ "python", "-u", "run.py" ]

37
api-cronjob/run.py
View File

@ -1,11 +1,15 @@
import os
import time
import requests
from datetime import datetime
from datetime import datetime, timezone, timedelta
import pytz
# Get the proxy URL from the environment variable
interval = 60*60*3 # 1 minute
update_cost = int(60*2)
scan_cost = int(10)
last_scan = None
scan_interval = 24*60*60
proxy_url = os.getenv('PROXY', "localhost")
user = os.getenv('ADMIN_USER_NAME', "")
password = os.getenv('ADMIN_PASSWORD', "")
@ -28,7 +32,33 @@ update_data = {
'subsidiary': None
}
"report_overview_duration"
# Define the scan API
scan_list_url = f'{proxy_url}/api/automation/'
scan_create_url = f'{proxy_url}/api/automation/(id)/scan/'
def semi_scan(login_token):
global last_scan
headers = {'Authorization': login_token}
sg_tz = sg_tz = pytz.timezone("Asia/Singapore")
# check if last scan is [scan_interval] ago
if not last_scan:
last_scan = time.time() - scan_interval
if time.time() - last_scan < scan_interval:
return
# get all rules:
list_rules_response = requests.get(scan_list_url, headers=headers)
print(f"[INFO]: Total {len(list_rules_response.json())} rules returned from server")
# process rule one by one
for rule in list_rules_response.json():
data = {
"start_date": datetime.now(sg_tz).strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "+08:00",
"end_date": (datetime.now(sg_tz) - timedelta(seconds=time.time()-last_scan)).strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "+08:00"
}
response = requests.post(scan_create_url.replace("(id)", str(rule["id"])), json=data, headers=headers)
print("[INFO]: scanning rule {} with data: {} status code: {}".format(rule["id"], data, response.status_code))
time.sleep(scan_cost)
last_scan = time.time()
# def update_report(login_token, report_overview_duration=["30d", "7d"], subsidiary=["all", "SEAU", "SESP", "SME", "SEPCO", "TSE", "SEIN"]):
def update_report(login_token, report_overview_duration=["7d", "30d"], subsidiary=["SEAO", "SEAU", "SESP", "SME", "SEPCO", "TSE", "SEIN"]):
@ -56,6 +86,7 @@ while True:
# Call the update API
try:
semi_scan(login_token)
update_report(login_token)
except Exception as e:
print(f"[ERROR]: {e}")