Api v2 delete note

pyproject.toml

@@ -1,3 +1,5 @@
 [tool.ruff.lint]
+preview = true
+select = ["E231", "E302", "E303", "E305", "E4", "E7", "E9", "F", "W292"]
 ignore = ["E402", "E711", "E712", "E721", "E722", "F821", "F841"]
 

source/app/__init__.py

@@ -19,7 +19,6 @@
 import json
 import logging as logger
 import os
-import urllib.parse
 from flask import Flask
 from flask import session
 from flask_bcrypt import Bcrypt
@@ -34,8 +33,11 @@
 from werkzeug.middleware.proxy_fix import ProxyFix
 
 from app.flask_dropzone import Dropzone
+from app.configuration import Config
 from app.iris_engine.tasker.celery import make_celery
+from app.iris_engine.tasker.celery import set_celery_flask_context
 from app.iris_engine.access_control.oidc_handler import get_oidc_client
+from app.jinja_filters import register_jinja_filters
 
 
 class ReverseProxied(object):
@@ -63,6 +65,15 @@ class AlertsNamespace(Namespace):
 
 logger.basicConfig(level=logger.INFO, format=LOG_FORMAT, datefmt=LOG_TIME_FORMAT)
 
+SQLALCHEMY_ENGINE_OPTIONS = {
+    "json_deserializer": partial(json.loads, object_pairs_hook=collections.OrderedDict),
+    "pool_pre_ping": True
+}
+
+db = SQLAlchemy(engine_options=SQLALCHEMY_ENGINE_OPTIONS)  # flask-sqlalchemy
+bc = Bcrypt()  # flask-bcrypt
+ma = Marshmallow()
+celery = make_celery(__name__)
 app = Flask(__name__, static_folder='../static')
 
 # CORS(app,
@@ -90,41 +101,30 @@ def ac_current_user_has_manage_perms():
     return False
 
 
-app.jinja_env.filters['unquote'] = lambda u: urllib.parse.unquote(u)
-app.jinja_env.filters['tojsonsafe'] = lambda u: json.dumps(u, indent=4, ensure_ascii=False)
-app.jinja_env.filters['tojsonindent'] = lambda u: json.dumps(u, indent=4)
-app.jinja_env.filters['escape_dots'] = lambda u: u.replace('.', '[.]')
+register_jinja_filters(app.jinja_env)
+
 app.jinja_env.globals.update(user_has_perm=ac_current_user_has_permission)
 app.jinja_env.globals.update(user_has_manage_perms=ac_current_user_has_manage_perms)
-app.jinja_options["autoescape"] = lambda _: True
+app.jinja_options['autoescape'] = lambda _: True
 app.jinja_env.autoescape = True
 
-app.config.from_object('app.configuration.Config')
-app.config.update(
-    SESSION_COOKIE_SECURE=True,
-    SESSION_COOKIE_HTTPONLY=True,
-    SESSION_COOKIE_SAMESITE='Lax'
-)
+app.config.from_object(Config)
+app.config['timezone'] = 'Europe/Paris'
+from app.post_init import PostInit
 
 cache = Cache(app)
 
-SQLALCHEMY_ENGINE_OPTIONS = {
-    "json_deserializer": partial(json.loads, object_pairs_hook=collections.OrderedDict),
-    "pool_pre_ping": True
-}
-
-db = SQLAlchemy(app, engine_options=SQLALCHEMY_ENGINE_OPTIONS)  # flask-sqlalchemy
+db.init_app(app)
 
-bc = Bcrypt(app)  # flask-bcrypt
+bc.init_app(app)
 
 lm = LoginManager()  # flask-loginmanager
 lm.init_app(app)  # init the login manager
-
-ma = Marshmallow(app) # Init marshmallow
+ma.init_app(app)
 
 dropzone = Dropzone(app)
 
-celery = make_celery(app)
+set_celery_flask_context(celery, app)
 
 # store = HttpExposedFileSystemStore(
 #     path='images',
@@ -140,13 +140,18 @@ def ac_current_user_has_manage_perms():
 socket_io.on_namespace(alerts_namespace)
 
 oidc_client = None
-if app.config.get('AUTHENTICATION_TYPE') == "oidc":
-    oidc_client = get_oidc_client(app)
+if app.config.get('AUTHENTICATION_TYPE') == 'oidc':
+    oidc_client = get_oidc_client(app.config, app.logger)
+from app.views import register_blueprints
+from app.views import load_user
+from app.views import load_user_from_request
+
 
 @app.teardown_appcontext
 def shutdown_session(exception=None):
     db.session.remove()
 
+
 @app.after_request
 def after_request(response):
     response.headers.add('Access-Control-Allow-Origin', '*')
@@ -157,17 +162,11 @@ def after_request(response):
     return response
 
 
-from app.views import register_blusprints
-from app.views import load_user
-from app.views import load_user_from_request
-
-register_blusprints(app)
-
-from app.post_init import run_post_init
+register_blueprints(app)
 
 try:
-
-    run_post_init(development=app.config['DEVELOPMENT'])
+    post_init = PostInit(app)
+    post_init.run()
 
 except Exception as e:
     app.logger.exception('Post init failed. IRIS not started')

source/app/alembic/alembic_utils.py

@@ -40,4 +40,4 @@ def index_exists(table_name, index_name):
     )
     inspector = reflection.Inspector.from_engine(engine)
     indexes = inspector.get_indexes(table_name)
-    return any(index['name'] == index_name for index in indexes)
+    return any(index['name'] == index_name for index in indexes)

source/app/alembic/env.py

@@ -1,7 +1,10 @@
+import os
 from alembic import context
 from logging.config import fileConfig
 from sqlalchemy import engine_from_config
 from sqlalchemy import pool
+from app.configuration import SQLALCHEMY_BASE_ADMIN_URI
+from app.configuration import PG_DB_
 
 # this is the Alembic Config object, which provides
 # access to the values within the .ini file in use.
@@ -11,11 +14,8 @@
 # This line sets up loggers basically.
 fileConfig(config.config_file_name)
 
-import os
 os.environ["ALEMBIC"] = "1"
 
-from app.configuration import SQLALCHEMY_BASE_ADMIN_URI, PG_DB_
-
 config.set_main_option('sqlalchemy.url', SQLALCHEMY_BASE_ADMIN_URI + PG_DB_)
 
 # add your model's MetaData object here

source/app/alembic/versions/3715d4fac4de_add_link_between_case_and_ioc.py

@@ -36,7 +36,6 @@ def upgrade():
         op.alter_column('ioc', 'case_id', nullable=True)
         return
 
-
     # Fetch all ioc_link rows
     ioc_links = conn.execute(text("SELECT ioc_id, case_id FROM ioc_link")).fetchall()
     if not ioc_links:
@@ -146,7 +145,6 @@ def upgrade():
         case_id = row.case_id
         events_by_ioc_and_case.setdefault((ioc_id, case_id), []).append(row.event_id)
 
-
     # We'll keep track of which (ioc_id, case_id) pairs we've already handled
     # so we don't do duplicate work if multiple ioc_link rows refer to the same pair.
     already_handled = set()
@@ -231,7 +229,6 @@ def upgrade():
             # Update our global in-memory map so future checks won't create a second duplicate
             existing_map[key] = new_ioc_id
 
-
             # Move ioc_comments to the new ioc
             if ioc_id in comments_by_ioc:
                 old_comment_links = comments_by_ioc[ioc_id]

source/app/alembic/versions/ff917e2ab02e_asset_type_custom_icons.py

@@ -80,7 +80,6 @@ def _get_icons(asset_name):
         "Windows Account - AD - krbtgt": ("Windows Account - AD - krbtgt", "user.png", "ioc_user.png"),
         "Windows Account - AD - Service": ("Windows Account - AD - krbtgt", "user.png", "ioc_user.png")
     }
-    if assets.get(asset_name):
-        return assets.get(asset_name)[1], assets.get(asset_name)[2]
-    else:
-        return "question-mark.png","ioc_question-mark.png"
+    if not assets.get(asset_name):
+        return "question-mark.png", "ioc_question-mark.png"
+    return assets.get(asset_name)[1], assets.get(asset_name)[2]

source/app/blueprints/graphql/cases.py

@@ -67,6 +67,7 @@ def resolve_case(root, info, case_id):
         permissions_check_current_user_has_some_case_access(case_id, [CaseAccessLevel.full_access])
         return Cases.query.get(case_id)
 
+
 class CaseConnection(Connection):
     class Meta:
         node = CaseObject

source/app/blueprints/graphql/graphql_route.py

@@ -55,8 +55,7 @@
 import warnings
 
 # Ignore all UserWarnings
-warnings.filterwarnings("ignore", category=UserWarning)
-
+warnings.filterwarnings('ignore', category=UserWarning)
 
 
 class Query(ObjectType):

source/app/blueprints/pages/case/case_notes_routes.py

@@ -58,7 +58,7 @@ def case_comment_note_modal(cur_id, caseid, url_redir):
     if url_redir:
         return redirect(url_for('case_note.case_note', cid=caseid, redirect=True))
 
-    note = get_note(cur_id, caseid=caseid)
+    note = get_note(cur_id)
     if not note:
         return response_error('Invalid note ID')
 

source/app/blueprints/pages/login/login_routes.py

@@ -36,9 +36,12 @@
 from app import bc
 from app import db
 from app import oidc_client
-from app.blueprints.access_controls import is_authentication_oidc, is_authentication_ldap
+from app.blueprints.access_controls import is_authentication_oidc
+from app.blueprints.access_controls import is_authentication_ldap
 from app.blueprints.responses import response_error
-from app.business.auth import validate_ldap_login, _retrieve_user_by_username, wrap_login_user
+from app.business.auth import validate_ldap_login
+from app.business.users import retrieve_user_by_username
+from app.business.auth import wrap_login_user
 from app.datamgmt.manage.manage_users_db import create_user
 from app.datamgmt.manage.manage_users_db import get_user
 from app.forms import LoginForm, MFASetupForm
@@ -56,7 +59,6 @@
 # filter User out of database through username
 
 
-
 def _render_template_login(form, msg):
     organisation_name = app.config.get('ORGANISATION_NAME')
     login_banner = app.config.get('LOGIN_BANNER_TEXT')
@@ -68,7 +70,7 @@ def _render_template_login(form, msg):
 
 
 def _validate_local_login(username, password):
-    user = _retrieve_user_by_username(username)
+    user = retrieve_user_by_username(username)
     if not user:
         return None
 
@@ -93,7 +95,7 @@ def _authenticate_ldap(form, username, password, local_fallback=True):
 
 
 def _authenticate_password(form, username, password):
-    user = _retrieve_user_by_username(username)
+    user = retrieve_user_by_username(username)
     if not user or user.is_service_account:
         return _render_template_login(form, 'Wrong credentials. Please try again.')
 
@@ -223,7 +225,7 @@ def oidc_authorise():
 
 @app.route('/auth/mfa-setup', methods=['GET', 'POST'])
 def mfa_setup():
-    user = _retrieve_user_by_username(username=session['username'])
+    user = retrieve_user_by_username(username=session['username'])
     form = MFASetupForm()
 
     if form.submit() and form.validate():
@@ -275,7 +277,7 @@ def mfa_verify():
 
         return redirect(url_for('login.login'))
 
-    user = _retrieve_user_by_username(username=session['username'])
+    user = retrieve_user_by_username(username=session['username'])
 
     # Redirect user to MFA setup if MFA is not fully set up
     if not user.mfa_secrets or not user.mfa_setup_complete:

source/app/blueprints/rest/alerts_routes.py

@@ -945,7 +945,6 @@ def alerts_batch_escalate_route() -> Response:
         return response_error(str(e))
 
 
-
 @alerts_rest_blueprint.route('/alerts/<int:alert_id>/comments/list', methods=['GET'])
 @ac_api_requires(Permissions.alerts_read)
 def alert_comments_get(alert_id):

source/app/blueprints/rest/case/case_assets_routes.py

@@ -397,4 +397,4 @@ def case_comment_asset_delete(cur_id, com_id, caseid):
     call_modules_hook('on_postload_asset_comment_delete', data=com_id, caseid=caseid)
 
     track_activity(f"comment {com_id} on asset {cur_id} deleted", caseid=caseid)
-    return response_success(msg)
+    return response_success(msg)