diff --git a/certificates.py b/certificates.py
index 8b850e73a205e5b90c0815fc8e897cf82adbd82d..58104069b04dfea635dd866a95522a8d34e4e653 100644
--- a/certificates.py
+++ b/certificates.py
@@ -22,29 +22,33 @@ import subprocess
import random
import string
from datetime import datetime, timedelta
+import smtplib
+import ssl
-
-import config
+from config import config
# Path to 'easyrsa' executable
#TODO: don't use the one from repository, should be installed somewhere
-EASYRSA = os.path.join(config.SOCTOOLS_BASE, "roles/ca/files/easyrsa/easyrsa")
+EASYRSA = config['easyrsa_bin']
# Environment variables to pass to easyrsa
EASYRSA_ENV = {
"EASYRSA_BATCH": "1",
- "EASYRSA_PKI": config.CA_DIR,
+ "EASYRSA_PKI": config['easyrsa_ca_dir'],
}
-TOKEN_FILE = os.path.join(config.SOCTOOLS_BASE, "secrets/cert_access_tokens")
+TOKEN_FILE = config['token_file']
TOKEN_FILE_HEADER = """
# Unique tokens allowing users to download the certificates generated for them via the user-mgmt-ui.
# !! AUTOMATICALLY GENERATED, DON'T EDIT !! (unless you know what you are doing)
""".lstrip()
TOKEN_EXPIRATION_HOURS = 24
-EMAIL_SUBJECT = "[SOCtools] New account for {name}"
-EMAIL_TEMPLATE = """
+EMAIL_TEMPLATE = """Content-Type: text/plain; charset=utf-8
+From: {sender}
+To: {recipient}
+Subject: [SOCtools] New account for {name}
+
Dear {name},
a user account has been created for you in the SOCtools system running at {base_url}.
@@ -52,13 +56,13 @@ a user account has been created for you in the SOCtools system running at {base_
Your username: {username}
You can authenticate to the system using a personal certificate which has been created for you.
-Please, download the certificate from the following link and install it into your browser:
+Please, download the certificate from the link below and install it into your OS or browser:
{access_url}
-Don't share the link with anyone! It allows to download your certificate, including the private key.
+Don't share the link with anyone! It allows to download your certificate and private key.
-The link will expire in {} hours.
+The link will expire in {expiration} hours.
"""
@@ -99,7 +103,7 @@ def export_p12_certificate(cn: str, password: str="") -> BinaryIO:
"""
_check_cn(cn)
# Export cert+key to a temporary file (using openssl, since easy-rsa doesn't allow to pass password on command line)
- out_file = os.path.join(config.CA_DIR, f"~tmp.{cn}.p12")
+ out_file = os.path.join(config['easyrsa_ca_dir'], f"~tmp.{cn}.p12")
crt_file, key_file = get_pem_files(cn)
cmd = ["openssl", "pkcs12", "-export", "-out", out_file, "-in", crt_file, "-inkey", key_file, "-passout", "pass:"+password]
print(f"Running command: {' '.join(cmd)}")
@@ -133,8 +137,8 @@ def get_pem_files(cn: str):
:return: Tuple with *.cert and *.key files in PEM format
"""
_check_cn(cn)
- return (os.path.join(config.CA_DIR, "issued", cn+".crt"),
- os.path.join(config.CA_DIR, "private", cn+".key"))
+ return (os.path.join(config['easyrsa_ca_dir'], "issued", cn+".crt"),
+ os.path.join(config['easyrsa_ca_dir'], "private", cn+".key"))
def generate_access_token(cn: str) -> str:
@@ -212,3 +216,46 @@ def _write_token_file(token_mapping: Dict[str, Tuple[str, datetime]]):
for token, (username, expiration) in token_mapping.items():
f.write(f"{token},{username},{expiration.strftime('%Y-%m-%dT%H:%M:%S')}\n")
+
+def send_token(user: 'UserAccount') -> Tuple[bool, Optional[str]]:
+ """
+ Generate token and send email to the user
+
+ :return tuple: success or not (bool), error message in case of failure
+ """
+ # Generate token
+ try:
+ token = generate_access_token(user.username)
+ except Exception as e:
+ return False, str(e)
+
+ access_url = f"{config['user_mgmt_base_url']}/export_certificate?token={token}"
+ # Print URL to console (for debugging or when email sending doesn't work)
+ print(f"Certificate access URL for '{user.username}': {access_url}")
+
+ # Send the token via email
+ name = f"{user.firstname} {user.lastname}".strip() or user.username
+ try:
+ _send_email(
+ config['smtp'],
+ user.email,
+ EMAIL_TEMPLATE.format(
+ sender=config['smtp']['sender'], recipient=user.email, name=name, username=user.username,
+ base_url=config['soctoolsproxy'], access_url=access_url, expiration=TOKEN_EXPIRATION_HOURS)
+ )
+ except Exception as e:
+ print(f"Error when trying to send email to the user: {e}")
+ return False, f"Error when trying to send email the user: {e}"
+
+ return True, None
+
+
+def _send_email(smtp_params: dict, to: str, message: str):
+ host = smtp_params['host']
+ port = smtp_params.get('port') or 465
+ sender = smtp_params['sender']
+ ssl_context = ssl.create_default_context()
+ with smtplib.SMTP_SSL(host, port, context=ssl_context) as smtp:
+ if smtp_params.get('username') and smtp_params.get('password'):
+ smtp.login(smtp_params['username'], smtp_params['password'])
+ smtp.sendmail(sender, to, message.encode('utf8'))
diff --git a/config.py b/config.py
index edfdb43c951d548781bcc36e68c430caefd17d6d..c7ed85fcd54a44d2680851ac97879a49143b9bfd 100644
--- a/config.py
+++ b/config.py
@@ -1,37 +1,4 @@
# Global configuration parameters
-# Some are filled during initialization by load_config() in main.py
-import os.path
-
-# *** Configuration of file paths ***
-SOCTOOLS_BASE = ".." # path to the root of soctools files
-VARIABLES_FILE = os.path.join(SOCTOOLS_BASE, "group_vars/all/variables.yml")
-CA_DIR = os.path.join(SOCTOOLS_BASE, "secrets/CA")
-CA_CERT_FILE = os.path.join(SOCTOOLS_BASE, "secrets/CA/ca.crt")
-KEYCLOAK_ADMIN_PASSWORD_FILE = os.path.join(SOCTOOLS_BASE, "secrets/passwords/keykloak_admin") # Note: should be keycloak, not keykloak
-
-MISP_API_KEY_FILE = os.path.join(SOCTOOLS_BASE, "secrets/tokens/misp")
-THEHIVE_API_KEY_FILE = os.path.join(SOCTOOLS_BASE, "secrets/tokens/thehive_secret_key")
-CORTEX_API_KEY_FILE = os.path.join(SOCTOOLS_BASE, "secrets/tokens/cortex_secret_key")
-
-# Credentials of the special user for account management
-# Cert and key should be in .pem format, unencrypted
-MGMT_USER_NAME = "soctools-user-mgmt"
-MGMT_USER_CERT_PATH = os.path.join(SOCTOOLS_BASE, "secrets/CA/issued/soctools-user-mgmt.crt")
-MGMT_USER_KEY_PATH = os.path.join(SOCTOOLS_BASE, "secrets/CA/private/soctools-user-mgmt.key")
-
-# Following parameters are set up dynamically by load_config() in main.py
-SOCTOOLSPROXY = None
-USER_MGMT_BASE_URL = None
-
-KEYCLOAK_BASE_URL = None
-KEYCLOAK_USERS_URL = None
-KEYCLOAK_ADMIN_PASSWORD = None
-
-MISP_API_KEY = None
-
-THEHIVE_API_KEY = None
-THEHIVE_ORG_NAME = None # set to "domain" from variables file
-
-CORTEX_API_KEY = None
-CORTEX_ORG_NAME = None # set to "domain" from variables file
+# The config parameters are loaded dynamically and stored into this:
+config = {}
diff --git a/config.yml.j2 b/config.yml.j2
new file mode 100644
index 0000000000000000000000000000000000000000..8052c7bbc3517bf85cf076024562e93e36bf92d9
--- /dev/null
+++ b/config.yml.j2
@@ -0,0 +1,45 @@
+# Global configuration parameters
+
+# Hostname of the SOCtools server
+soctoolsproxy: "{{soctoolsproxy}}"
+
+# SMTP connection parameters (used to send emails with access information to users)
+smtp:
+ # hostname of SMTP server to use
+ host: "{{smtp.host}}"
+ #port: 465
+ # sender email address ("From:" header)
+ sender: "{{smtp.sender}}"
+ # user and pass to authenticate (optional, it tries to send email without authentication if empty)
+ username: "{{smtp.username}}"
+ password: "{{smtp.password}}"
+
+# Path to the SOCtools CA certificate
+ca_cert_file: "../secrets/CA/ca.crt"
+
+# Path to "easyrsa" executable and working directory
+easyrsa_bin: "../roles/ca/files/easyrsa/easyrsa"
+easyrsa_ca_dir: "../secrets/CA"
+
+# File to store tokens allowing users to download certificates
+token_file: "../secrets/cert_access_tokens"
+
+# Credentials of the special user for account management
+# Cert and key should be in .pem format, unencrypted
+mgmt_user_name: "soctools-user-mgmt"
+mgmt_user_cert_path: "../secrets/CA/issued/soctools-user-mgmt.crt"
+mgmt_user_key_path: "../secrets/CA/private/soctools-user-mgmt.key"
+
+user_mgmt_base_url: "https://{{soctoolsproxy}}:5443"
+
+keycloak_base_url: "https://{{soctoolsproxy}}:12443"
+keycloak_users_url: "https://{{soctoolsproxy}}:12443/auth/admin/realms/{{openid_realm}}/users"
+keycloak_admin_password: "{{lookup('password', '{{playbook_dir}}/secrets/passwords/keycloak_admin')}}"
+
+misp_api_key: "{{lookup('password', '{{playbook_dir}}/secrets/tokens/misp')}}"
+
+thehive_api_key: "{{lookup('password', '{{playbook_dir}}/secrets/tokens/thehive_secret_key')}}"
+thehive_org_name: "{{org_name}}"
+
+cortex_api_key: "{{lookup('password', '{{playbook_dir}}/secrets/tokens/cortex_secret_key')}}"
+cortex_org_name: "{{org_name}}"
diff --git a/cortex.py b/cortex.py
index dbbaf15a7810a69f9b219ab06c76fd5cb635c876..947721cbd01958ac794ec6c61115eb167a1dcd79 100644
--- a/cortex.py
+++ b/cortex.py
@@ -4,7 +4,7 @@ from typing import List, Dict, Optional
import requests
from datetime import datetime
-import config
+from config import config
# Base URL to Cortex API endpoints
CORTEX_API_BASE_URL = "https://{soctools_proxy}:9001/api"
@@ -35,9 +35,9 @@ def cortex_get_users() -> List[Dict]:
:raise CortexUnexpectedReplyError
"""
# all users in given org
- url = CORTEX_API_BASE_URL.format(soctools_proxy=config.SOCTOOLSPROXY) + f"/organization/{config.THEHIVE_ORG_NAME}/user"
+ url = CORTEX_API_BASE_URL.format(soctools_proxy=config['soctoolsproxy']) + f"/organization/{config['thehive_org_name']}/user"
# all users
- #url = CORTEX_API_BASE_URL.format(soctools_proxy=config.SOCTOOLSPROXY) + f"/user"
+ #url = CORTEX_API_BASE_URL.format(soctools_proxy=config['soctoolsproxy']) + f"/user"
resp = _send_request("get", url)
if not resp.ok:
@@ -66,11 +66,11 @@ def cortex_add_user(user: 'UserAccount') -> None:
:raise CortexUnexpectedReplyError
"""
- url = CORTEX_API_BASE_URL.format(soctools_proxy=config.SOCTOOLSPROXY) + "/user"
+ url = CORTEX_API_BASE_URL.format(soctools_proxy=config['soctoolsproxy']) + "/user"
data = {
"login": user.email,
"name": f"{user.firstname} {user.lastname}".strip() or user.username,
- "organization": config.CORTEX_ORG_NAME,
+ "organization": config['cortex_org_name'],
"roles": ["read", "analyze"], #, "orgadmin"],
}
resp = _send_request("post", url, data)
@@ -91,10 +91,10 @@ def cortex_edit_user(login: str, user: 'UserAccount') -> None:
:raise CortexUserNotFoundError,CortexUnexpectedReplyError
"""
- url = CORTEX_API_BASE_URL.format(soctools_proxy=config.SOCTOOLSPROXY) + f"/user/{login}"
+ url = CORTEX_API_BASE_URL.format(soctools_proxy=config['soctoolsproxy']) + f"/user/{login}"
data = {
"name": f"{user.firstname} {user.lastname}".strip() or user.username,
- #"organisation": config.CORTEX_ORG_NAME,
+ #"organisation": config['cortex_org_name'],
#"roles": ["read", "analyze"], # TODO allow to set different roles?
}
print(url, data)
@@ -115,7 +115,7 @@ def cortex_delete_user(login: str) -> None:
:raise CortexUserNotFoundError,CortexUnexpectedReplyError
"""
- url = CORTEX_API_BASE_URL.format(soctools_proxy=config.SOCTOOLSPROXY) + f"/user/{login}"
+ url = CORTEX_API_BASE_URL.format(soctools_proxy=config['soctoolsproxy']) + f"/user/{login}"
data = {
"status": "Locked"
}
@@ -135,8 +135,8 @@ def _send_request(method:str, url:str, data:Optional[dict]=None):
return getattr(requests, method)(
url,
headers={
- "Authorization": "Bearer " + config.CORTEX_API_KEY,
+ "Authorization": "Bearer " + config['cortex_api_key'],
},
- verify=config.CA_CERT_FILE,
+ verify=config['ca_cert_file'],
json=data
)
diff --git a/main.py b/main.py
index fe7709984baf37123878d6eadef7904ca4dd6db6..21c8cdb3f0a87677e8c2518b81d0816853c0fb7d 100644
--- a/main.py
+++ b/main.py
@@ -8,11 +8,55 @@ from flask_wtf import FlaskForm
from wtforms import StringField, BooleanField
from wtforms.validators import DataRequired, Email
-import requests
import yaml
from dataclasses import dataclass, field # external package - backport of Py3.7 dataclasses to Py3.6
+import re
+
+from config import config # module serving as a global namespace for configuration parameters
+
+
+# Load and check configuration - paths, URLs, api keys, etc.
+invalid_configuration_error = None
+try:
+ config.update(yaml.safe_load(open("config.yml", "r")))
+ errors = []
+
+ # Check presence of all mandatory configuration parameters
+ for attr in ('soctoolsproxy', 'ca_cert_file', 'easyrsa_bin', 'easyrsa_ca_dir', 'token_file',
+ 'mgmt_user_name', 'mgmt_user_cert_path', 'mgmt_user_key_path', 'user_mgmt_base_url',
+ 'keycloak_base_url', 'keycloak_users_url', 'keycloak_admin_password',
+ 'misp_api_key', 'thehive_api_key', 'thehive_org_name', 'cortex_api_key', 'cortex_org_name',):
+ if not config[attr]:
+ errors.append(f'Missing mandatory parameter "{attr}".')
+
+ # Check soctoolsproxy
+ if not re.match('[a-zA-Z0-9.:-]+', config['soctoolsproxy']):
+ errors.append("'soctoolsproxy' is not a valid hostname or IP address.")
+
+ # Check smtp params, set defaults
+ if not isinstance(config.get('smtp'), dict):
+ errors.append('Missing mandatory parameter "smtp" or it is not dict.')
+ if not re.match('[a-zA-Z0-9.:-]+', config['smtp'].get('host')):
+ errors.append('Missing mandatory parameter "smtp.host" or is not a valid hostname or IP address.')
+ if not config['smtp'].get('sender'):
+ errors.append('Missing mandatory parameter "smtp.sender".')
+ if not config['smtp'].get('port'):
+ config['smtp']['port'] = 465
+
+ if errors:
+ invalid_configuration_error = "Configuration error(s):\n" + "\n".join(errors)
+except Exception as e:
+ invalid_configuration_error = f"Can't load configuration: {e}"
+
+
+print(f"Config loaded:\nsoctoolsproxy={config['soctoolsproxy']}\nkeycloak_base_url={config['keycloak_base_url']}\n"
+ f"keycloak_admin_password={config['keycloak_admin_password'][:3]}...{config['keycloak_admin_password'][-4:]}\n"
+ f"misp_api_key={config['misp_api_key'][:3]}...{config['misp_api_key'][-4:]}\n"
+ f"thehive_api_key={config['thehive_api_key'][:3]}...{config['thehive_api_key'][-4:]}\n"
+ f"cortex_api_key={config['cortex_api_key'][:3]}...{config['cortex_api_key'][-4:]}\n"
+ f"thehive_org_name={config['thehive_org_name']}\n")
+
-import config
import certificates
from nifi import *
from misp import *
@@ -22,39 +66,17 @@ from cortex import *
app = Flask(__name__)
app.secret_key = "ASDF1234 - CHANGE ME!" # TODO: set dynamically to something random
+# If there is config error, report it instead of trying to render a page
+@app.before_request
+def config_check():
+ if invalid_configuration_error is not None:
+ return make_response(f"500 Internal Server Error\n{'='*25}\n\n{invalid_configuration_error}\n\n"
+ "Fix the configuration file and restart the user-mgmt-ui service "
+ "('systemctl restart user-mgmt-ui')", 500, {'Content-Type': 'text/plain'})
-@app.before_first_request
-def load_config():
- """Load various variables, api keys, etc. and set configuration parameters"""
- variables = yaml.safe_load(open(config.VARIABLES_FILE, "r"))
- # Get FQDN of the main server
- config.SOCTOOLSPROXY = variables["soctoolsproxy"]
- assert re.match('[a-zA-Z0-9.-]+', config.SOCTOOLSPROXY), f"ERROR: The 'soctoolsproxy' variable loaded from '{config.VARIABLES_FILE}' is not a valid domain name."
- # Set base URL for user management (this web service)
- # TODO: load ports (or whole base URLs) from config as well
- config.USER_MGMT_BASE_URL = f"http://{config.SOCTOOLSPROXY}:8080"
- # Set base URL to Keycloak
- config.KEYCLOAK_BASE_URL = f"https://{config.SOCTOOLSPROXY}:12443"
- config.KEYCLOAK_USERS_URL = config.KEYCLOAK_BASE_URL + "/auth/admin/realms/SOCTOOLS1/users"
- # Load API key for Keycloak
- config.KEYCLOAK_ADMIN_PASSWORD = open(config.KEYCLOAK_ADMIN_PASSWORD_FILE, "r").read(100).strip() # read max 100 B, the key should never be so long
- # Load API key for MISP
- config.MISP_API_KEY = open(config.MISP_API_KEY_FILE, "r").read(100).strip() # read max 100 B, the key should never be so long
- # Load API key for The Hive
- config.THEHIVE_API_KEY = open(config.THEHIVE_API_KEY_FILE, "r").read(100).strip() # read max 100 B, the key should never be so long
- # Load API key for Cortex
- config.CORTEX_API_KEY = open(config.CORTEX_API_KEY_FILE, "r").read(100).strip() # read max 100 B, the key should never be so long
- # Load organization name (for The Hive and Cortex)
- config.THEHIVE_ORG_NAME = variables["org_name"]
- config.CORTEX_ORG_NAME = variables["org_name"]
-
- print(f"Config loaded:\nSOCTOOLSPROXY={config.SOCTOOLSPROXY}\nKEYCLOAK_BASE_URL={config.KEYCLOAK_BASE_URL}\n"
- f"KEYCLOAK_ADMIN_PASSWORD={config.KEYCLOAK_ADMIN_PASSWORD[:3]}...{config.KEYCLOAK_ADMIN_PASSWORD[-4:]}\n"
- f"MISP_API_KEY={config.MISP_API_KEY[:3]}...{config.MISP_API_KEY[-4:]}\n"
- f"THEHIVE_API_KEY={config.THEHIVE_API_KEY[:3]}...{config.THEHIVE_API_KEY[-4:]}\n"
- f"CORTEX_API_KEY={config.CORTEX_API_KEY[:3]}...{config.CORTEX_API_KEY[-4:]}\n"
- f"THEHIVE_ORG_NAME={config.THEHIVE_ORG_NAME}\n")
+def redirect_to_main_page():
+ return redirect(config['user_mgmt_base_url'] + "/")
# *** Custom Jinja filters ***
def ts_to_str(ts: Union[float,datetime,None]) -> str:
@@ -122,15 +144,15 @@ def kc_get_token() -> str:
Return the token or raise KeycloakError
"""
- url = config.KEYCLOAK_BASE_URL + "/auth/realms/master/protocol/openid-connect/token"
+ url = config['keycloak_base_url'] + "/auth/realms/master/protocol/openid-connect/token"
data = {
"client_id": "admin-cli",
"username": "admin",
- "password": config.KEYCLOAK_ADMIN_PASSWORD,
+ "password": config['keycloak_admin_password'],
"grant_type": "password"
}
try:
- resp = requests.post(url, data, verify=config.CA_CERT_FILE)
+ resp = requests.post(url, data, verify=config['ca_cert_file'])
if resp.status_code != 200:
raise KeycloakError(f"Can't get OIDC token for API access: ({resp.status_code}) {resp.text[:200]}")
return str(resp.json()['access_token'])
@@ -145,7 +167,7 @@ def kc_get_users() -> List[UserAccount]:
:raise KeycloakError
"""
token = kc_get_token()
- resp = requests.get(config.KEYCLOAK_USERS_URL, headers={'Authorization': 'Bearer ' + token}, verify=config.CA_CERT_FILE)
+ resp = requests.get(config['keycloak_users_url'], headers={'Authorization': 'Bearer ' + token}, verify=config['ca_cert_file'])
if not resp.ok:
raise KeycloakError(f"Can't get list of users: ({resp.status_code}) {resp.text[:200]}")
try:
@@ -165,8 +187,8 @@ def kc_get_user_by_id(userid: str) -> UserAccount:
"""
assert re.match(r'[0-9a-z-]*', userid), "Invalid user ID"
token = kc_get_token()
- url = config.KEYCLOAK_USERS_URL + "/" + userid
- resp = requests.get(url, headers={'Authorization': 'Bearer ' + token}, verify=config.CA_CERT_FILE)
+ url = config['keycloak_users_url'] + "/" + userid
+ resp = requests.get(url, headers={'Authorization': 'Bearer ' + token}, verify=config['ca_cert_file'])
if not resp.ok:
raise KeycloakError(f"Can't get user info: ({resp.status_code}) {resp.text[:200]}")
try:
@@ -185,8 +207,8 @@ def kc_get_user_by_name(username: str) -> Optional[UserAccount]:
:raise KeycloakError
"""
token = kc_get_token()
- url = config.KEYCLOAK_USERS_URL
- resp = requests.get(url, params={'username': username, 'exact': 'true'}, headers={'Authorization': 'Bearer ' + token}, verify=config.CA_CERT_FILE)
+ url = config['keycloak_users_url']
+ resp = requests.get(url, params={'username': username, 'exact': 'true'}, headers={'Authorization': 'Bearer ' + token}, verify=config['ca_cert_file'])
if not resp.ok:
raise KeycloakError(f"Can't get user info: ({resp.status_code}) {resp.text[:200]}")
print(resp.text)
@@ -210,8 +232,8 @@ def kc_add_user(user: UserAccount) -> None:
user_data = user.to_keycloak_representation()
user_data["enabled"] = True # add "enable" key, since a new user must be explicitly enabled (default is False)
- resp = requests.post(config.KEYCLOAK_USERS_URL, json=user_data,
- headers={'Authorization': 'Bearer ' + token}, verify=config.CA_CERT_FILE)
+ resp = requests.post(config['keycloak_users_url'], json=user_data,
+ headers={'Authorization': 'Bearer ' + token}, verify=config['ca_cert_file'])
if not resp.ok:
raise KeycloakError(f"({resp.status_code}) {resp.text[:200]}")
@@ -226,9 +248,9 @@ def kc_update_user(user: UserAccount) -> None:
token = kc_get_token()
user_data = user.to_keycloak_representation()
- url = config.KEYCLOAK_USERS_URL + "/" + user.kcid
+ url = config['keycloak_users_url'] + "/" + user.kcid
resp = requests.put(url, json=user_data,
- headers={'Authorization': 'Bearer ' + token}, verify=config.CA_CERT_FILE)
+ headers={'Authorization': 'Bearer ' + token}, verify=config['ca_cert_file'])
if not resp.ok:
raise KeycloakError(f"({resp.status_code}) {resp.text[:200]}")
@@ -241,8 +263,8 @@ def kc_delete_user(userid: str) -> None:
"""
assert re.match(r'[0-9a-z-]*', userid), "Invalid user ID"
token = kc_get_token()
- url = config.KEYCLOAK_USERS_URL + "/" + userid
- resp = requests.delete(url, headers={'Authorization': 'Bearer ' + token}, verify=config.CA_CERT_FILE)
+ url = config['keycloak_users_url'] + "/" + userid
+ resp = requests.delete(url, headers={'Authorization': 'Bearer ' + token}, verify=config['ca_cert_file'])
if not resp.ok:
raise KeycloakError(f"({resp.status_code}) {resp.text[:200]}")
@@ -275,7 +297,7 @@ def main():
users = []
# Mark "internal" users
for u in users:
- if u.username == config.MGMT_USER_NAME:
+ if u.username == config['mgmt_user_name']:
u.internal = True
#print(users)
@@ -288,7 +310,7 @@ def main():
nifi_users = []
# Mark "internal" users
for u in nifi_users:
- if u["name"].startswith("CN=soctools-nifi-") or u["name"] == config.MGMT_USER_NAME:
+ if u["name"].startswith("CN=soctools-nifi-") or u["name"] == config['mgmt_user_name']:
u["internal"] = True
# List of usernames only (for easier cross-check with Keycloak users)
nifi_usernames = set(nu["name"] for nu in nifi_users)
@@ -356,7 +378,7 @@ def add_user():
except certificates.CertError as e:
flash(str(e), "error")
- return redirect("/") # don't continue creating user accounts in services
+ return redirect_to_main_page() # don't continue creating user accounts in services
# Keycloak
try:
@@ -364,7 +386,7 @@ def add_user():
flash(f'User "{user.username}" successfully created in Keycloak.', "success")
except Exception as e:
flash(f'Error when creating user in Keycloak: {e}', "error")
- return redirect("/") # don't continue creating user accounts in other services
+ return redirect_to_main_page() # don't continue creating user accounts in other services
# NiFi
try:
@@ -404,13 +426,13 @@ def add_user():
# Send email to the user
if form_user.send_email.data:
- ok, err = _send_token(user.username, user.email) # TODO
+ ok, err = certificates.send_token(user)
if ok:
flash(f"Email successfully sent to '{user.email}'", "success")
else:
flash(f"ERROR: {err}", "error")
- return redirect("/") # Success - go back to main page
+ return redirect_to_main_page() # Success - go back to main page
return render_template("add_edit_user.html", form_user=form_user, user=None)
@@ -422,7 +444,7 @@ def edit_user(username: str):
user = kc_get_user_by_name(username)
except KeycloakError as e:
flash(f'ERROR: {e}', "error")
- return redirect('/')
+ return redirect_to_main_page()
keycloak_id = user.kcid
# POST = perform the update
if request.method == "POST":
@@ -491,7 +513,7 @@ def edit_user(username: str):
except Exception as e:
flash(f'Error when updating user in Cortex: {e}', "error")
- return redirect("/") # Success - go back to main page
+ return redirect_to_main_page() # Success - go back to main page
# data not valid - show form again
return render_template("add_edit_user.html", form_user=form_user, user={"kcid": keycloak_id})
@@ -507,7 +529,7 @@ def delete_user(username: str):
user_spec = kc_get_user_by_name(username)
except KeycloakError as e:
flash(f"Error: Can't get user info from KeyCloak: {e}", "error")
- return redirect("/")
+ return redirect_to_main_page()
# TODO revoke certificate
@@ -554,7 +576,7 @@ def delete_user(username: str):
except TheHiveError as e:
flash(f'Error when trying to mark user as "locked" in Cortex: {e}', "error")
- return redirect("/")
+ return redirect_to_main_page()
@app.route("/export_certificate")
@@ -577,7 +599,7 @@ def export_certificate():
user_spec = kc_get_user_by_name(username)
if user_spec is None:
flash(f"ERROR: No such user ('{username}')", "error")
- return redirect("/")
+ return redirect_to_main_page()
# If format is given, export and serve the certificate file
format = request.args.get("format", None)
@@ -592,7 +614,7 @@ def export_certificate():
return send_file(certificates.get_pem_files(user_spec.cn)[1],
attachment_filename=f"{user_spec.cn}.key", mimetype="application/x-pem-file")
# Otherwise show the HTML page
- return render_template("export_certificate.html", token=token, username=username, soctoolsproxy=config.SOCTOOLSPROXY)
+ return render_template("export_certificate.html", token=token, username=username, soctoolsproxy=config['soctoolsproxy'])
@app.route("/send_token/<username>")
@@ -606,35 +628,15 @@ def send_token(username: str):
user_spec = kc_get_user_by_name(username)
if user_spec is None:
flash(f"ERROR: No such user ('{username}')", "error")
- return redirect("/")
+ return redirect_to_main_page()
- ok, err = _send_token(username, user_spec.email)
+ ok, err = certificates.send_token(user_spec)
if ok:
flash(f"Email successfully sent to '{user_spec.email}'", "success")
else:
flash(f"ERROR: {err}", "error")
- return redirect("/")
-
-
-def _send_token(username: str, email: str) -> Tuple[bool, Optional[str]]:
- """
- Generate token and send the email (internal function)
-
- :return tuple: success or not (bool), error message in case of failure
- """
- # Generate token
- try:
- token = certificates.generate_access_token(username)
- except Exception as e:
- return False, str(e)
-
- access_url = f"{config.USER_MGMT_BASE_URL}/export_certificate?token={token}"
- print(f"Certificate access URL for '{username}': {access_url}")
-
- # Send the token via email
- # TODO
+ return redirect_to_main_page()
- return True, None
# TODO:
@@ -642,11 +644,11 @@ def _send_token(username: str, email: str) -> Tuple[bool, Optional[str]]:
# - send tokens via email
# - authentication/authorization to this GUI
-# @app.route("/test_cert/<func>")
-# def test_cert_endpoint(func):
-# # run any function from "certificates" module
-# result = str(getattr(certificates, func)(**request.args))
-# return make_response(result)
+@app.route("/test_cert/<func>")
+def test_cert_endpoint(func):
+ # run any function from "certificates" module
+ result = str(getattr(certificates, func)(**request.args))
+ return make_response(result)
# When the script is run directly, run the application on a local development server.
# Optionally pass two parameters, 'host' (IP to listen on) and 'port',
diff --git a/misp.py b/misp.py
index d5b99da4f69b3d6c0b6a0e0e2d1585ba0a316e10..c5fe5af3c56626c45e206c2b5f2c61c22a77fdea 100644
--- a/misp.py
+++ b/misp.py
@@ -3,11 +3,8 @@
from typing import List, Dict, Optional
import requests
from datetime import datetime
-import re
-from operator import itemgetter
-import urllib.parse
-import config
+from config import config
# Base URL to MISP API endpoints
MISP_API_BASE_URL = "https://{soctools_proxy}:6443"
@@ -37,7 +34,7 @@ def misp_get_users() -> List[Dict]:
:return List of dicts with keys 'id', 'email', 'org', 'role', 'login', 'created' (datetime), 'last_login' (datetime or None)
:raise MISPUnexpectedReplyError
"""
- url = MISP_API_BASE_URL.format(soctools_proxy=config.SOCTOOLSPROXY) + "/admin/users"
+ url = MISP_API_BASE_URL.format(soctools_proxy=config['soctoolsproxy']) + "/admin/users"
resp = _send_request("get", url)
if not resp.ok:
@@ -73,7 +70,7 @@ def misp_add_user(user: 'UserAccount') -> None:
user_role_id = 1 # should be "admin", no support for other roles yet
user_org_id = 1 # use the first org, no support for selection yet
- url = MISP_API_BASE_URL.format(soctools_proxy=config.SOCTOOLSPROXY) + "/admin/users/add"
+ url = MISP_API_BASE_URL.format(soctools_proxy=config['soctoolsproxy']) + "/admin/users/add"
data = {
"email": user_email,
"org_id": user_org_id,
@@ -99,7 +96,7 @@ def misp_edit_user(old_email, new_email) -> None:
"""
user_id = _get_id_by_email(old_email) # raises MISPUserNotFoundError if user not found
- url = MISP_API_BASE_URL.format(soctools_proxy=config.SOCTOOLSPROXY) + "/admin/users/edit/" + user_id
+ url = MISP_API_BASE_URL.format(soctools_proxy=config['soctoolsproxy']) + "/admin/users/edit/" + user_id
data = {
"email": new_email,
"external_auth_required": "1",
@@ -122,7 +119,7 @@ def misp_delete_user(user_email: str) -> None:
:raises MISPUnexpectedReplyError, MISPUserNotFoundError
"""
user_id = _get_id_by_email(user_email) # raises MISPUserNotFoundError if user not found
- url = MISP_API_BASE_URL.format(soctools_proxy=config.SOCTOOLSPROXY) + "/admin/users/delete/" + user_id
+ url = MISP_API_BASE_URL.format(soctools_proxy=config['soctoolsproxy']) + "/admin/users/delete/" + user_id
resp = _send_request("post", url)
if not resp.ok:
print(f"Can't delete user from MISP: Unexpected reply {resp.status_code}: {resp.text[:500]}")
@@ -137,16 +134,16 @@ def _send_request(method:str, url:str, data:Optional[dict]=None):
return getattr(requests, method)(
url,
headers={
- "Authorization": config.MISP_API_KEY,
+ "Authorization": config['misp_api_key'],
"Accept": "application/json",
},
- verify=config.CA_CERT_FILE,
+ verify=config['ca_cert_file'],
json=data
)
def _get_id_by_email(email:str) -> str:
- url = MISP_API_BASE_URL.format(soctools_proxy=config.SOCTOOLSPROXY) + "/admin/users"
+ url = MISP_API_BASE_URL.format(soctools_proxy=config['soctoolsproxy']) + "/admin/users"
resp = _send_request("get", url)
if not resp.ok:
raise MISPUnexpectedReplyError(f"Can't find id of user with email '{email}': Unexpected reply {resp.status_code}")
diff --git a/nifi.py b/nifi.py
index 1cb446bf377bd137c721cfc1f48933b8983624a2..63c4454cd11e85ce1c1242dcded7c738452fb8cc 100644
--- a/nifi.py
+++ b/nifi.py
@@ -6,7 +6,7 @@ import re
from operator import itemgetter
import urllib.parse
-import config
+from config import config
# URL to initial login process
NIFI_LOGIN_URL = "https://{soctools_proxy}:9443/nifi/login"
@@ -38,7 +38,7 @@ def nifi_get_users() -> List[Dict]:
:return List of dicts with keys 'id', 'name', 'groups' (list of group names)
:raise NifiUnexpectedReplyError
"""
- url = NIFI_API_BASE_URL.format(soctools_proxy=config.SOCTOOLSPROXY) + "/tenants/users"
+ url = NIFI_API_BASE_URL.format(soctools_proxy=config['soctoolsproxy']) + "/tenants/users"
token = _nifi_get_jwt()
resp = _send_request('get', url, token)
if not resp.ok:
@@ -109,14 +109,14 @@ def _nifi_get_jwt() -> str:
# We will need to store some cookies - create a session which will automatically handle it
# Also, set path to certificates
session = requests.Session()
- session.verify = config.CA_CERT_FILE
- session.cert = (config.MGMT_USER_CERT_PATH, config.MGMT_USER_KEY_PATH)
+ session.verify = config['ca_cert_file']
+ session.cert = (config['mgmt_user_cert_path'], config['mgmt_user_key_path'])
# Initiate login process by querying the NiFi login page.
# NiFi should set the 'nifi-oidc-request-identifier' cookie (stored into session, will be needed later) and
# redirect us to Keycloak. The redirection is automatically followed by requests.get().
# Keycloak should authenticate us using the provided certificate and present a web page with confirmation form.
- url = NIFI_LOGIN_URL.format(soctools_proxy=config.SOCTOOLSPROXY)
+ url = NIFI_LOGIN_URL.format(soctools_proxy=config['soctoolsproxy'])
resp = session.get(url, allow_redirects=True)
if not resp.ok:
raise NifiUnexpectedReplyError(f"_nifi_get_jwt: Received unexpected HTTP status code ({resp.status_code}) from URL '{url}'.")
@@ -143,7 +143,7 @@ def _nifi_get_jwt() -> str:
# Now, we are authenticated to NiFi, identified by a cookie (stored within our session object).
# Use the cookie and ask for the JWT token we need for API requests
- url = NIFI_API_BASE_URL.format(soctools_proxy=config.SOCTOOLSPROXY) + "/access/oidc/exchange"
+ url = NIFI_API_BASE_URL.format(soctools_proxy=config['soctoolsproxy']) + "/access/oidc/exchange"
resp = session.post(url) # POST must be used even though no data are being sent
if not resp.ok:
raise NifiUnexpectedReplyError(f"_nifi_get_jwt: Received unexpected HTTP status code ({resp.status_code}) from URL '{url}'.")
@@ -154,7 +154,7 @@ def _send_request(method:str, url:str, token:str, data:Optional[dict]=None):
return getattr(requests, method)(
url,
headers={"Authorization": "Bearer " + token},
- verify=config.CA_CERT_FILE,
+ verify=config['ca_cert_file'],
json=data
)
@@ -163,7 +163,7 @@ def _get_group_by_name(group_name: str, token=None) -> Optional[dict]:
"""Return complete user-group specification of a group with given name (or None if not found)"""
if not token:
token = _nifi_get_jwt()
- url = NIFI_API_BASE_URL.format(soctools_proxy=config.SOCTOOLSPROXY) + "/tenants/user-groups"
+ url = NIFI_API_BASE_URL.format(soctools_proxy=config['soctoolsproxy']) + "/tenants/user-groups"
resp = _send_request('get', url, token)
if not resp.ok:
raise NifiUnexpectedReplyError(f"Can't get list of user groups from NiFi: Unexpected reply {resp.status_code}")
@@ -200,7 +200,7 @@ def _add_user_to_group(user_id: str, group_name: str, token=None):
group["component"]["users"].append({"id": user_id})
# Save new group
- url = NIFI_API_BASE_URL.format(soctools_proxy=config.SOCTOOLSPROXY) + "/tenants/user-groups/" + group_id
+ url = NIFI_API_BASE_URL.format(soctools_proxy=config['soctoolsproxy']) + "/tenants/user-groups/" + group_id
resp = _send_request('put', url, token, data=group)
if not resp.ok:
raise NifiUnexpectedReplyError(f"Can't assign user to group in NiFi: Unexpected reply: {resp.status_code} {resp.text[:1000]}")
@@ -209,7 +209,7 @@ def _add_user_to_group(user_id: str, group_name: str, token=None):
def _add_user(user_name: str, token=None):
if not token:
token = _nifi_get_jwt()
- url = NIFI_API_BASE_URL.format(soctools_proxy=config.SOCTOOLSPROXY) + "/tenants/users"
+ url = NIFI_API_BASE_URL.format(soctools_proxy=config['soctoolsproxy']) + "/tenants/users"
post_data = {
"revision": {"version": 0},
"component": {"identity": user_name}
@@ -227,7 +227,7 @@ def _get_user_by_name(user_name: str, token=None) -> Optional[dict]:
"""Get user specification of the user with given name (or None if no such user found)"""
if not token:
token = _nifi_get_jwt()
- url = NIFI_API_BASE_URL.format(soctools_proxy=config.SOCTOOLSPROXY) + "/tenants/search-results?q=" + urllib.parse.quote(user_name)
+ url = NIFI_API_BASE_URL.format(soctools_proxy=config['soctoolsproxy']) + "/tenants/search-results?q=" + urllib.parse.quote(user_name)
resp = _send_request('get', url, token)
if not resp.ok:
raise NifiUnexpectedReplyError(f"Can't get user info from NiFi: Unexpected reply: {resp.status_code} {resp.text[:1000]}")
@@ -244,7 +244,7 @@ def _delete_user(user_spec: dict, token=None):
if not token:
token = _nifi_get_jwt()
- url = NIFI_API_BASE_URL.format(soctools_proxy=config.SOCTOOLSPROXY) + "/tenants/users/" + user_spec["id"]
+ url = NIFI_API_BASE_URL.format(soctools_proxy=config['soctoolsproxy']) + "/tenants/users/" + user_spec["id"]
url += "?version=" + str(int(user_spec["revision"]["version"]))
print("DELETE " + url)
resp = _send_request('delete', url, token)
diff --git a/thehive.py b/thehive.py
index 2083b029abade974713df3143bc17a1e8d0c7f7b..2f32d2eb91d1ec3c1e5d7a4efab617c38b6d7cfe 100644
--- a/thehive.py
+++ b/thehive.py
@@ -4,7 +4,7 @@ from typing import List, Dict, Optional
import requests
from datetime import datetime
-import config
+from config import config
# Base URL to The Hive API endpoints
THEHIVE_API_BASE_URL = "https://{soctools_proxy}:9000/api/v1"
@@ -35,10 +35,10 @@ def thehive_get_users() -> List[Dict]:
:return List of dicts with keys 'id', 'login', 'name', 'role', 'org', 'created' (datetime)
:raise TheHiveUnexpectedReplyError
"""
- url = THEHIVE_API_BASE_URL.format(soctools_proxy=config.SOCTOOLSPROXY) + "/query"
+ url = THEHIVE_API_BASE_URL.format(soctools_proxy=config['soctoolsproxy']) + "/query"
data = {
"query": [
- {"_name": "getOrganisation", "idOrName": config.THEHIVE_ORG_NAME},
+ {"_name": "getOrganisation", "idOrName": config['thehive_org_name']},
{"_name": "users"}
]
}
@@ -73,11 +73,11 @@ def thehive_add_user(user: 'UserAccount') -> None:
:raise TheHiveUnexpectedReplyError
"""
- url = THEHIVE_API_BASE_URL.format(soctools_proxy=config.SOCTOOLSPROXY) + "/user"
+ url = THEHIVE_API_BASE_URL.format(soctools_proxy=config['soctoolsproxy']) + "/user"
data = {
"login": user.email,
"name": f"{user.firstname} {user.lastname}".strip() or user.username,
- "organisation": config.THEHIVE_ORG_NAME,
+ "organisation": config['thehive_org_name'],
"profile": "analyst", # TODO allow to set different roles?
#"email": user.email,
#"password": "",
@@ -97,10 +97,10 @@ def thehive_edit_user(login: str, user: 'UserAccount') -> None:
:raise TheHiveUnexpectedReplyError
"""
- url = THEHIVE_API_BASE_URL.format(soctools_proxy=config.SOCTOOLSPROXY) + f"/user/{login}"
+ url = THEHIVE_API_BASE_URL.format(soctools_proxy=config['soctoolsproxy']) + f"/user/{login}"
data = {
"name": f"{user.firstname} {user.lastname}".strip() or user.username,
- #"organisation": config.THEHIVE_ORG_NAME,
+ #"organisation": config['thehive_org_name'],
#"profile": "analyst", # TODO allow to set different roles?
}
@@ -119,7 +119,7 @@ def thehive_delete_user(login: str) -> None:
:raise TheHiveUnexpectedReplyError
"""
- url = THEHIVE_API_BASE_URL.format(soctools_proxy=config.SOCTOOLSPROXY) + f"/user/{login}/force?organisation={config.THEHIVE_ORG_NAME}"
+ url = THEHIVE_API_BASE_URL.format(soctools_proxy=config['soctoolsproxy']) + f"/user/{login}/force?organisation={config['thehive_org_name']}"
#print(url)
resp = _send_request("delete", url)
#print(resp.text)
@@ -138,8 +138,8 @@ def _send_request(method:str, url:str, data:Optional[dict]=None):
return getattr(requests, method)(
url,
headers={
- "Authorization": "Bearer " + config.THEHIVE_API_KEY,
+ "Authorization": "Bearer " + config['thehive_api_key'],
},
- verify=config.CA_CERT_FILE,
+ verify=config['ca_cert_file'],
json=data
)