Flask路径扫描工具问题：仅返回首个有效URL及后台执行需求

Hey there! Let's fix those two issues with your Flask directory scanner one by one. I'll walk you through practical, actionable solutions for both problems.

1. Returning All Valid URLs (Not Just the First Match) #

The root problem here is your dir function uses return as soon as it finds a 200-status URL, which exits the loop immediately. Instead, we need to collect all valid URLs in a list and return the full list after checking every path. Here's how to update your code:

Updated dir.py #

import requests

def scan_dirs(domain):
    paths = ['/test', '/html', '/go', '/.git']
    valid_urls = []
    for path in paths:
        path = path.strip()
        url = f'http://{domain}{path}'
        try:
            # Add a timeout to avoid hanging on unresponsive endpoints
            response = requests.get(url, timeout=5)
            if response.status_code == 200:
                valid_urls.append(url)
        except requests.exceptions.RequestException:
            # Skip any paths that cause connection errors, timeouts, etc.
            continue
    return valid_urls

Then, in your run.py, make sure you're capturing this list and passing it to your template properly:

# Replace brute_admin with scan_dirs (assuming brute_admin was calling your old dir function)
brute_res = scan_dirs(domain)
return render_template("brute_result.html", brute_res=brute_res)

In your brute_result.html, loop through brute_res to display all valid URLs:

{% if brute_res %}
    <h3>Valid URLs Found:</h3>
    <ul>
        {% for url in brute_res %}
            <li>{{ url }}</li>
        {% endfor %}
    </ul>
{% else %}
    <p>No valid URLs found.</p>
{% endif %}

2. Running Scans in the Background & Enabling Result Downloads #

For long-running scans, we need to offload the task to a background worker so users don't have to wait. The most robust solution for Flask is using Celery with a message broker like Redis (it's easy to set up and works well for small to medium apps). Here's how to implement this:

Step 1: Install Required Packages #

First, install Celery and Redis:

pip install celery redis

Make sure Redis is running locally (you can download it directly or use a container for convenience).

Step 2: Configure Celery in Your Flask App #

Update your run.py to set up Celery:

from flask import Flask, request, render_template, redirect, url_for
from celery import Celery
from dir import scan_dirs  # Import our updated scan function

app = Flask(__name__)

# Celery configuration
app.config['CELERY_BROKER_URL'] = 'redis://localhost:6379/0'
app.config['CELERY_RESULT_BACKEND'] = 'redis://localhost:6379/0'

def make_celery(app):
    celery = Celery(
        app.import_name,
        broker=app.config['CELERY_BROKER_URL'],
        backend=app.config['CELERY_RESULT_BACKEND']
    )
    celery.conf.update(app.config)
    return celery

celery = make_celery(app)

# Convert scan_dirs to an async Celery task
@celery.task
def async_scan_dirs(domain):
    return scan_dirs(domain)

Step 3: Update the Brute View to Handle Async Tasks #

Modify your /brute route to submit the scan as a background task and redirect the user to a status page:

@app.route('/brute', methods=["GET", "POST"])
def brute():
    erroro = ""
    if request.method == "POST":
        domain_input = request.form.get("domain", "").strip()
        if not domain_input:
            erroro = "<p>Domain input cannot be empty</p>"
        else:
            # Submit the scan task to Celery
            task = async_scan_dirs.delay(domain_input)
            # Redirect to task status page with the task ID
            return redirect(url_for('task_status', task_id=task.id))
    
    return render_template('brute.html', erroro=erroro)

Step 4: Add Task Status & Download Endpoints #

Add routes to let users check scan progress and download results once done:

from flask import Response

@app.route('/task/<task_id>')
def task_status(task_id):
    task = async_scan_dirs.AsyncResult(task_id)
    
    if task.state == 'PENDING':
        status_data = {
            'state': task.state,
            'message': 'Scanning in progress... Please wait.'
        }
    elif task.state == 'SUCCESS':
        status_data = {
            'state': task.state,
            'valid_urls': task.result,
            'task_id': task_id
        }
    else:
        status_data = {
            'state': task.state,
            'message': f"Error occurred: {str(task.info)}"
        }
    
    return render_template('task_status.html', status=status_data)

@app.route('/download/<task_id>')
def download_results(task_id):
    task = async_scan_dirs.AsyncResult(task_id)
    
    if task.state != 'SUCCESS':
        return "Scan is still in progress or failed. Please check the task status first.", 400
    
    valid_urls = task.result
    # Create a plain text file with all valid URLs
    file_content = "\n".join(valid_urls) if valid_urls else "No valid URLs found."
    
    # Return the file as a download
    return Response(
        file_content,
        mimetype="text/plain",
        headers={
            "Content-Disposition": f"attachment; filename=valid_urls_{task_id}.txt"
        }
    )

Step 5: Create the Task Status Template (task_status.html) #

This template will show scan progress and a download button when done:

<!DOCTYPE html>
<html>
<head>
    <title>Scan Status</title>
</head>
<body>
    <h2>Scan Status: {{ status.state }}</h2>
    {% if status.message %}
        <p>{{ status.message }}</p>
    {% endif %}
    
    {% if status.valid_urls %}
        <h3>Valid URLs Found:</h3>
        <ul>
            {% for url in status.valid_urls %}
                <li>{{ url }}</li>
            {% endfor %}
        </ul>
        <a href="{{ url_for('download_results', task_id=status.task_id) }}">Download Results</a>
    {% endif %}
    
    {% if status.state == 'PENDING' %}
        <p>Refresh this page to check progress.</p>
    {% endif %}
</body>
</html>

Step 6: Run the Celery Worker #

Finally, start the Celery worker in a separate terminal window:

celery -A run.celery worker --loglevel=info

Then run your Flask app as usual. Now users can submit a domain, get redirected to a status page, and download the results once the scan finishes in the background.

内容的提问来源于stack exchange，提问作者Rudra Sarkar