Merge branch 'adrima01-simple_interface'

bin/async_capture.py

@@ -116,7 +116,8 @@ class AsyncCapture(AbstractManager):
 
                 if send_report:
                     self.lookyloo.send_mail(uuid, email=settings.get('email', ''),
-                                            comment=settings.get('comment'))
+                                            comment=settings.get('comment'),
+                                            recipient_mail=settings.get("recipient_mail"))
 
             lazy_cleanup = self.lookyloo.redis.pipeline()
             if queue and self.lookyloo.redis.zscore('queues', queue):

config/users/admin.json.sample

@@ -0,0 +1,7 @@
+{
+  "overwrite": true,
+  "listing": false,
+  "auto_report": {
+    "recipient_mail": "analyst@test.de"
+  }
+}

lookyloo/default/helpers.py

@@ -57,6 +57,10 @@ def load_configs(path_to_config_files: str | Path | None=None) -> None:
     for path in config_path.glob('*.json'):
         with path.open() as _c:
             configs[path.stem] = json.load(_c)
+    user_path = config_path / 'users'
+    for path in user_path.glob('*.json'):
+        with path.open() as _c:
+            configs[path.stem] = json.load(_c)
 
 
 @lru_cache(64)

lookyloo/helpers.py

@@ -29,6 +29,7 @@ from werkzeug.utils import cached_property
 
 from .default import get_homedir, safe_create_dir, get_config, LookylooException
 
+
 logger = logging.getLogger('Lookyloo - Helpers')
 
 

lookyloo/lookyloo.py

@@ -16,6 +16,7 @@ import time
 
 from collections import defaultdict
 from datetime import date, datetime, timedelta, timezone
+from functools import lru_cache
 from email.message import EmailMessage
 from functools import cached_property
 from io import BytesIO
@@ -78,6 +79,22 @@ class CaptureSettings(CaptureSettingsCore, total=False):
     parent: str | None
 
 
+# overwrite set to True means the settings in the config file overwrite the settings
+# provided by the user. False will simply append the settings from the config file if they
+# don't exist.
+class UserCaptureSettings(CaptureSettings, total=False):
+    overwrite: bool
+
+
+@lru_cache(64)
+def load_user_config(username: str) -> UserCaptureSettings | None:
+    user_config_path = get_homedir() / 'config' / 'users' / f'{username}.json'
+    if not user_config_path.exists():
+        return None
+    with user_config_path.open() as _c:
+        return json.load(_c)
+
+
 class Lookyloo():
 
     def __init__(self, cache_max_size: int | None=None) -> None:
@@ -629,6 +646,26 @@ class Lookyloo():
                 query['document'] = document
         return query
 
+    def _apply_user_config(self, query: CaptureSettings, user_config: UserCaptureSettings) -> CaptureSettings:
+        def recursive_merge(dict1: CaptureSettings | UserCaptureSettings,
+                            dict2: CaptureSettings | UserCaptureSettings) -> CaptureSettings:
+            # dict2 overwrites dict1
+            for key, value in dict2.items():
+                if key in dict1 and isinstance(dict1[key], dict) and isinstance(value, dict):  # type: ignore[literal-required]
+                    # Recursively merge nested dictionaries
+                    dict1[key] = recursive_merge(dict1[key], value)  # type: ignore[literal-required,arg-type]
+                else:
+                    # Merge non-dictionary values
+                    dict1[key] = value  # type: ignore[literal-required]
+            return dict1
+
+        # merge
+        if user_config.pop('overwrite', None):
+            # config from file takes priority
+            return recursive_merge(query, user_config)
+        else:
+            return recursive_merge(user_config, query)
+
     def enqueue_capture(self, query: CaptureSettings, source: str, user: str, authenticated: bool) -> str:
         '''Enqueue a query in the capture queue (used by the UI and the API for asynchronous processing)'''
 
@@ -652,6 +689,9 @@ class Lookyloo():
                 query[key] = json.dumps(value) if value else None  # type: ignore[literal-required]
 
         query = self._prepare_lacus_query(query)
+        if authenticated:
+            if user_config := load_user_config(user):
+                query = self._apply_user_config(query, user_config)
 
         priority = get_priority(source, user, authenticated)
         if priority < -100:
@@ -864,7 +904,8 @@ class Lookyloo():
 
         return f"Malicious capture according to {len(modules)} module(s): {', '.join(modules)}"
 
-    def send_mail(self, capture_uuid: str, /, email: str='', comment: str | None=None) -> bool | dict[str, Any]:
+    def send_mail(self, capture_uuid: str, /, email: str='', comment: str | None=None,
+                  recipient_mail: str | None = None) -> bool | dict[str, Any]:
         '''Send an email notification regarding a specific capture'''
         if not get_config('generic', 'enable_mail_notification'):
             return {"error": "Unable to send mail: mail notification disabled"}
@@ -913,7 +954,7 @@ class Lookyloo():
         msg['From'] = email_config['from']
         if email:
             msg['Reply-To'] = email
-        msg['To'] = email_config['to']
+        msg['To'] = email_config['to'] if not recipient_mail else recipient_mail
         msg['Subject'] = email_config['subject']
         body = get_email_template()
         body = body.format(

website/web/__init__.py

@@ -1588,7 +1588,7 @@ def capture_web() -> str | Response | WerkzeugResponse:
         if request.form.get('proxy'):
             parsed_proxy = urlparse(request.form['proxy'])
             if parsed_proxy.scheme and parsed_proxy.hostname and parsed_proxy.port:
-                if parsed_proxy.scheme in ['http', 'https', 'socks5']:
+                if parsed_proxy.scheme in ['http', 'https', 'socks5', 'socks5h']:
                     if (parsed_proxy.username and parsed_proxy.password) or (not parsed_proxy.username and not parsed_proxy.password):
                         capture_query['proxy'] = request.form['proxy']
                     else:
@@ -1637,6 +1637,38 @@ def capture_web() -> str | Response | WerkzeugResponse:
     return _prepare_capture_template(user_ua=request.headers.get('User-Agent'))
 
 
+@app.route('/simple_capture', methods=['GET', 'POST'])
+@flask_login.login_required  # type: ignore[misc]
+def simple_capture() -> str | Response | WerkzeugResponse:
+    user = flask_login.current_user.get_id()
+    if request.method == 'POST':
+        if not (request.form.get('url') or request.form.get('urls')):
+            flash('Invalid submission: please submit at least a URL.', 'error')
+            return render_template('simple_capture.html')
+        capture_query: CaptureSettings = {}
+        if request.form.get('url'):
+            capture_query['url'] = request.form['url']
+            perma_uuid = lookyloo.enqueue_capture(capture_query, source='web', user=user,
+                                                  authenticated=flask_login.current_user.is_authenticated)
+            time.sleep(2)
+            if perma_uuid:
+                flash('Recording is in progress and is reported automatically.', 'success')
+            return redirect(url_for('simple_capture'))
+        elif request.form.get('urls'):
+            for url in request.form['urls'].strip().split('\n'):
+                if not url:
+                    continue
+                query = capture_query.copy()
+                query['url'] = url
+                new_capture_uuid = lookyloo.enqueue_capture(query, source='web', user=user,
+                                                            authenticated=flask_login.current_user.is_authenticated)
+                if new_capture_uuid:
+                    flash('Recording is in progress and is reported automatically.', 'success')
+            return redirect(url_for('simple_capture'))
+    # render template
+    return render_template('simple_capture.html')
+
+
 @app.route('/cookies/<string:cookie_name>', methods=['GET'])
 def cookies_name_detail(cookie_name: str) -> str:
     captures, domains = get_cookie_name_investigator(cookie_name.strip())

website/web/helpers.py

@@ -5,6 +5,7 @@ from __future__ import annotations
 import hashlib
 import json
 import os
+import re
 from functools import lru_cache
 from pathlib import Path
 
@@ -50,6 +51,10 @@ def load_user_from_request(request: Request) -> User | None:
     return None
 
 
+def is_valid_username(username: str) -> bool:
+    return bool(re.match("^[A-Za-z0-9]+$", username))
+
+
 @lru_cache(64)
 def build_keys_table() -> dict[str, str]:
     keys_table = {}
@@ -72,6 +77,9 @@ def get_users() -> dict[str, str | list[str]]:
 def build_users_table() -> dict[str, dict[str, str]]:
     users_table: dict[str, dict[str, str]] = {}
     for username, authstuff in get_users().items():
+        if not is_valid_username(username):
+            raise Exception('Invalid username, can only contain characters and numbers.')
+
         if isinstance(authstuff, str):
             # just a password, make a key
             users_table[username] = {}

website/web/templates/index.html

@@ -83,6 +83,11 @@ $(document).ready(function () {
     <a href="{{ url_for('submit_capture') }}">
       <button class="new-capture-button btn btn-primary">Submit capture</button>
     </a>
+    {% if current_user.is_authenticated %}
+    <a href="{{ url_for('simple_capture') }}">
+      <button class="new-capture-button btn btn-primary">Takedown process</button>
+    </a>
+    {% endif %}
     <br><br>
     {{ render_messages(container=True, dismissible=True) }}
   </center>

website/web/templates/simple_capture.html

@@ -0,0 +1,103 @@
+{% extends "main.html" %}
+{% from 'bootstrap5/utils.html' import render_messages %}
+{% block title %}Capture{% endblock %}
+
+{% block card %}
+<meta property="og:title" content="Lookyloo" />
+<meta property="og:type" content="website"/>
+<meta
+  property="og:description"
+  content="Lookyloo captures websites and let you investigate them."
+/>
+<meta
+  property="og:image"
+  content="https://{{public_domain}}{{ url_for('static', filename='lookyloo.jpeg') }}"
+/>
+<meta
+  property="og:url"
+  content="https://{{public_domain}}"
+/>
+<meta name="twitter:card" content="summary_large_image">
+{% endblock %}
+
+{% block content %}
+<div class="container">
+  <center>
+    <a href="{{ url_for('index') }}" title="Go back to index">
+      <img src="{{ url_for('static', filename='lookyloo.jpeg') }}"
+           alt="Lookyloo" width="25%">
+    </a>
+  </center>
+  {{ render_messages(container=True, dismissible=True) }}
+  <form role="form" action="{{ url_for('simple_capture') }}" method=post enctype=multipart/form-data>
+    <!-- Submission type -->
+
+    <div class="tab-content" id="nav-tabContent">
+      </br>
+      <div class="tab-pane fade show active" id="nav-url" role="tabpanel" aria-labelledby="nav-url-tab">
+        <div class="row input-group mb-3">
+          <label for="singleCaptureField" class="col-sm-1 col-form-label">URL(s):</label>
+          <input type="text" class="form-control col-auto" name="url" id=singleCaptureField
+                 placeholder="URL to capture" value="{{predefined_url_to_capture}}" required>
+
+          <textarea class="form-control col-auto d-none" placeholder="URLs to capture, one per line"
+                    name="urls" id=multipleCapturesField></textarea>
+
+          <span class="col-sm-2 input-group-text">
+            <div class="form-check">
+              <input class="form-check-input" name="multipleCaptures" id="multipleCaptures" type="checkbox"
+                     value="" aria-label="tick to enable multiple captures">
+              <label for="multipleCaptures" class="form-check-label">Multiple captures</label>
+            </div>
+          </span>
+        </div>
+      </div>
+    </div>
+    <hr>
+
+    <center>
+      </br>
+      <button type="submit" class="new-capture-button btn btn-primary" id="btn-looking">Submit!</button>
+    </center>
+  </form>
+</div>
+{% endblock %}
+
+{% block scripts %}
+  {{ super() }}
+  <script src='{{ url_for('static', filename='capture.js') }}'
+          integrity="{{get_sri('static', 'capture.js')}}"
+          crossorigin="anonymous"></script>
+  <script>
+  $('#nav-url-tab').on('click', function(e) {
+    document.getElementById("singleCaptureField").required = true;
+    document.getElementById("document").required = false;
+    $("#singleCaptureField").removeClass("d-none");
+    document.getElementById('multipleCaptures').checked = false;
+    $("#multipleCapturesField").addClass("d-none");
+  });
+  $('#nav-doc-tab').on('click', function(e) {
+    document.getElementById("document").required = true;
+    document.getElementById("multipleCapturesField").required = false;
+    document.getElementById("singleCaptureField").required = false;
+  });
+  </script>
+  <script>
+      $('#multipleCaptures').on('click', function(e) {
+        if (document.getElementById('multipleCaptures').checked == true) {
+            document.getElementById('singleCaptureField').value = '';
+            $("#singleCaptureField").addClass("d-none");
+            document.getElementById("singleCaptureField").required = false;
+            $("#multipleCapturesField").removeClass("d-none");
+            document.getElementById("multipleCapturesField").required = true;
+        }
+        else {
+            document.getElementById('multipleCapturesField').value = '';
+            $("#singleCaptureField").removeClass("d-none");
+            document.getElementById("singleCaptureField").required = true;
+            $("#multipleCapturesField").addClass("d-none");
+            document.getElementById("multipleCapturesField").required = false;
+        }
+      })
+  </script>
+{% endblock %}