RustImplementation #23
+29
-5
@@ -12,7 +12,7 @@
|
||||
#
|
||||
# You should have received a copy of the GNU Affero General Public License
|
||||
# along with this program. If not, see <https://www.gnu.org/licenses/>.
|
||||
|
||||
import argparse
|
||||
import dotenv
|
||||
import os
|
||||
import pandas as pd
|
||||
@@ -20,6 +20,7 @@ import urllib3
|
||||
import utils.allowlist
|
||||
import utils.getdeviceevents
|
||||
import utils.hashfunctions
|
||||
import utils.otpfunctions
|
||||
import utils.pathfunctions
|
||||
import utils.policyfunctions
|
||||
import utils.pretty as ct
|
||||
@@ -38,12 +39,36 @@ path_exclusion_constant = 4
|
||||
min_files_for_path = 4
|
||||
threat_tolerance_constant = 4
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description="Your script description")
|
||||
parser.add_argument('--monitorOTP', action='store_true', help='Run in non-interactive mode')
|
||||
# Add other arguments as needed
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
if args.monitorOTP:
|
||||
# Non-interactive logic
|
||||
apivalidation()
|
||||
print(f"Running non-interactively to start monitoring OTP")
|
||||
|
||||
utils.otpfunctions.getOTPHistory(url, 0)
|
||||
utils.otpfunctions.getOTPHistory(url, 1)
|
||||
utils.otpfunctions.getOTPHistory(url, 2)
|
||||
utils.otpfunctions.getOTPHistory(url, 3)
|
||||
|
||||
# Process input here
|
||||
else:
|
||||
# Interactive logic
|
||||
apivalidation()
|
||||
menu_main()
|
||||
|
||||
|
||||
def apivalidation():
|
||||
match os.getenv('APIKEY'):
|
||||
case '':
|
||||
print(ct.colorText("Please add your API Key to the .env file", "red"))
|
||||
case _:
|
||||
menu_main()
|
||||
|
||||
|
||||
def tryToReadCSV(csv):
|
||||
try:
|
||||
@@ -253,5 +278,4 @@ def menu_prepare_to_enforce():
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
apivalidation()
|
||||
|
||||
main()
|
||||
@@ -0,0 +1,21 @@
|
||||
import json
|
||||
import requests
|
||||
import os
|
||||
|
||||
|
||||
|
||||
def getOTPHistory(url, type):
|
||||
|
||||
endpoint = url + '/v1/otp/usage'
|
||||
payload = {
|
||||
"status" : type
|
||||
}
|
||||
|
||||
headers = {"X-APIKey": os.getenv('APIKEY')}
|
||||
payload = json.dumps(payload)
|
||||
|
||||
response = requests.post(endpoint, headers=headers, data=payload, verify=False)
|
||||
data = response.json()
|
||||
results = data.get("response", {}).get("otpusage", [])
|
||||
|
||||
print(results)
|
||||
@@ -24,13 +24,13 @@ import utils.allowlist
|
||||
|
||||
|
||||
|
||||
def addHash(policy, hash):
|
||||
def addHash(url, policy, hash):
|
||||
print(f"Adding the following hashes to {policy}:")
|
||||
for p in hash:
|
||||
print(p)
|
||||
|
||||
|
||||
def addPath(policy, hash):
|
||||
def addPath(url, policy, hash):
|
||||
print(f"Adding the following Path Exclusions to {policy}:")
|
||||
for p in hash:
|
||||
print(p)
|
||||
|
||||
Reference in New Issue
Block a user