This is an internal network information collection tool developed based on PyQt6, capable of automatically scanning various text files (including code files, configuration files, log files, etc.) in a specified directory. It uses customizable regular expression scanning rules to detect and identify sensitive information in files, such as passwords, API keys, database connection strings, and other security risks. The tool supports multi-threaded concurrent processing to improve scanning efficiency, provides real-time progress monitoring, and generates detailed HTML format scanning reports. Users can flexibly configure the activation status and risk levels of detection rules according to actual needs, focusing on internal network security audits and sensitive information leakage detection.
Main Features
File Scanning and Information Collection
- Automatically scans text files in the specified directory to find potential sensitive information
- Supports various file formats: .txt, .py, .js, .json, .xml, .yaml, .sql, .log, .md, .html, .php, .java, etc.
- Uses regular expression pattern matching to identify sensitive content
Multi-threaded Processing
- Supports concurrent scanning with 2-16 threads to improve scanning efficiency
- Configurable thread count to adapt to different system performances
Intelligent File Filtering
- Automatically skips system folders (such as .git, node_modules, __pycache__, etc.)
- Supports various character encodings (UTF-8, GBK, GB2312, Latin-1)
The tool is built using PyQt6 and includes the following features:
1. Scanning
- Directory Selection: Choose the target directory to scan
- Scanning Options: Configure the number of threads and report file name
- Progress Monitoring: Real-time display of scanning progress and currently processed files
- Action Buttons: Start scan, stop scan, view report
2. Results
- Statistics: Displays total number of files, scanned files, skipped files, error files, and risk files
- Results Table: Summarizes found sensitive information by type
- Detailed View: Double-click to view specific matching results and file locations
3. Configuration
- Rule Management: View and manage detection rules
- Rule Switch: Enable or disable specific detection rules
- Configuration Editing: Supports editing configuration files
Technical Features
Pattern Matching System
- Detection rules are stored in a JSON configuration file
- Each rule includes a regular expression, risk level, and description
- Supports dynamic loading and reloading of configurations
Report Generation
- Automatically generates detailed scanning reports in HTML format
- Includes statistics, risk classifications, and specific matching results
- Supports viewing reports in a browser
Applicable Scenarios
- Internal Network Security Check: Detect sensitive information leakage in code and configuration files
- Compliance Audit: Ensure sensitive data meets security standards
- Code Review: Check for potential security risks before code submission
- Data Classification: Identify and classify different types of sensitive information
Code
import sys
import os
import re
import json
import threading
from datetime import datetime
from pathlib import Path
from typing import Dict, List, Tuple, Optional
import mimetypes
from PyQt6.QtGui import QIcon
from concurrent.futures import ThreadPoolExecutor, as_completed
from PyQt6.QtWidgets import (QApplication, QMainWindow, QWidget, QVBoxLayout, QHBoxLayout, QGridLayout, QTabWidget, QGroupBox, QLabel, QLineEdit, QPushButton, QProgressBar, QTextEdit, QTableWidget, QTableWidgetItem, QFileDialog, QMessageBox, QComboBox, QCheckBox, QHeaderView, QFrame)
from PyQt6.QtCore import Qt, QThread, pyqtSignal
from PyQt6.QtGui import QFont
class PatternLoader:
def __init__(self):
self.config_file = "patterns.json"
self.patterns = self.load_patterns()
def load_patterns(self):
if not os.path.exists(self.config_file):
return {}
try:
with open(self.config_file, 'r', encoding='utf-8') as f:
data = json.load(f)
return data
except Exception as e:
print(f"Failed to load configuration file: {e}")
return {}
def get_active_patterns(self):
compiled = {}
for name, config in self.patterns.items():
if config.get("enabled", True):
try:
compiled[name] = re.compile(config["regex"], re.MULTILINE | re.DOTALL)
except re.error:
continue
return compiled
class ScanThread(QThread):
progress_signal = pyqtSignal(int, str)
status_signal = pyqtSignal(str)
finished_signal = pyqtSignal(dict, dict)
def __init__(self, directory, patterns, threads):
super().__init__()
self.directory = directory
self.patterns = patterns
self.thread_count = threads
self.stop_flag = False
self.results = {}
self.stats = {"total": 0, "scanned": 0, "skipped": 0, "errors": 0}
def run(self):
try:
self.status_signal.emit("Collecting files...")
files = self.collect_files()
self.stats["total"] = len(files)
if not files:
self.status_signal.emit("No files found")
return
self.status_signal.emit(f"Starting scan of {len(files)} files")
self.process_files(files)
if not self.stop_flag:
self.status_signal.emit("Scan complete")
self.finished_signal.emit(self.results, self.stats)
except Exception as e:
self.status_signal.emit(f"Scan error: {str(e)}")
def collect_files(self):
files = []
text_exts = {'.txt', '.py', '.js', '.json', '.xml', '.yaml', '.yml', '.ini', '.cfg', '.conf', '.properties', '.sql', '.log', '.md', '.html', '.htm', '.css', '.php', '.java', '.cpp', '.c', '.h', '.cs', '.go', '.rs', '.rb', '.pl', '.sh'}
for root, dirs, filenames in os.walk(self.directory):
dirs[:] = [d for d in dirs if not d.startswith('.') and d not in {'node_modules', '__pycache__', '.git'}]
for filename in filenames:
if filename.startswith('.'):
continue
file_path = os.path.join(root, filename)
ext = Path(filename).suffix.lower()
if ext in text_exts:
files.append(file_path)
return files
def process_files(self, files):
with ThreadPoolExecutor(max_workers=self.thread_count) as executor:
futures = {executor.submit(self.scan_file, f): f for f in files}
completed = 0
for future in as_completed(futures):
if self.stop_flag:
break
file_path = futures[future]
completed += 1
try:
result = future.result()
if result:
self.results[file_path] = result
self.stats["scanned"] += 1
except Exception:
self.stats["errors"] += 1
progress = int((completed / len(files)) * 100)
filename = os.path.basename(file_path)
self.progress_signal.emit(progress, filename)
def scan_file(self, file_path):
try:
content = self.read_file(file_path)
if not content:
return {}
results = {}
lines = content.split('\n')
for name, pattern in self.patterns.items():
matches = []
for line_num, line in enumerate(lines, 1):
for match in pattern.finditer(line):
matches.append({
'text': match.group(0),
'line': line_num
})
if matches:
results[name] = matches
return results
except Exception:
return {}
def read_file(self, file_path):
encodings = ['utf-8', 'gbk', 'gb2312', 'latin-1']
for encoding in encodings:
try:
with open(file_path, 'r', encoding=encoding) as f:
return f.read()
except UnicodeDecodeError:
continue
except Exception:
break
return ""
def stop(self):
self.stop_flag = True
Reply in the backgroundInformation collectionGet the complete source code and tool