Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
64 changes: 55 additions & 9 deletions backend/Generator/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,8 @@
import os
import fitz
import mammoth
import zipfile
from werkzeug.utils import secure_filename

class MCQGenerator:

Expand Down Expand Up @@ -368,19 +370,63 @@ def extract_text_from_docx(self, file_path):
return result.value

def process_file(self, file):
file_path = os.path.join(self.upload_folder, file.filename)
if not file.filename:
raise ValueError("Empty filename")

filename = secure_filename(file.filename)
if not filename:
raise ValueError("Invalid filename")

target_dir = os.path.abspath(self.upload_folder)
file_path = os.path.abspath(os.path.join(target_dir, filename))

if os.path.commonpath([target_dir, file_path]) != target_dir:
raise ValueError("Path traversal detected")

file.save(file_path)
content = ""

if file.filename.endswith('.txt'):
with open(file_path, 'r') as f:
content = f.read()
elif file.filename.endswith('.pdf'):
content = self.extract_text_from_pdf(file_path)
elif file.filename.endswith('.docx'):
content = self.extract_text_from_docx(file_path)
try:
if '.' not in filename:
raise ValueError("Unsupported file extension")
ext = filename.rsplit('.', 1)[1].lower()

with open(file_path, 'rb') as f:
header = f.read(512)

if ext == 'pdf':
if not header.startswith(b"%PDF"):
raise ValueError("Invalid file content: PDF signature not found")
content = self.extract_text_from_pdf(file_path)

elif ext == 'docx':
if not header.startswith(b"PK\x03\x04"):
raise ValueError("Invalid file content: DOCX signature not found")

try:
with zipfile.ZipFile(file_path, 'r') as zf:
namelist = zf.namelist()
if '[Content_Types].xml' not in namelist or 'word/document.xml' not in namelist:
raise ValueError("Invalid file content: Missing DOCX internal structures")
except zipfile.BadZipFile:
raise ValueError("Invalid file content: Not a valid ZIP archive")

content = self.extract_text_from_docx(file_path)
Comment thread
coderabbitai[bot] marked this conversation as resolved.

elif ext == 'txt':
try:
header.decode('utf-8')
except UnicodeDecodeError:
raise ValueError("Invalid file content: TXT must be valid UTF-8 text")
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read()
Comment on lines +416 to +422

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major

Validate the full text file, not just the first 512 bytes.

Lines 407-410 decode only the header. A file with valid UTF-8 in the first 512 bytes and invalid bytes later will pass this check and then fail during f.read(), bypassing the intended ValueError path. Decode the full payload once and raise ValueError on any UnicodeDecodeError.

Suggested fix
             elif ext == 'txt':
-                try:
-                    header.decode('utf-8')
-                except UnicodeDecodeError:
-                    raise ValueError("Invalid file content: TXT must be valid UTF-8 text")
-                with open(file_path, 'r', encoding='utf-8') as f:
-                    content = f.read()
+                with open(file_path, 'rb') as f:
+                    raw = f.read()
+                try:
+                    content = raw.decode('utf-8')
+                except UnicodeDecodeError as exc:
+                    raise ValueError("Invalid file content: TXT must be valid UTF-8 text") from exc
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
elif ext == 'txt':
try:
header.decode('utf-8')
except UnicodeDecodeError:
raise ValueError("Invalid file content: TXT must be valid UTF-8 text")
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read()
elif ext == 'txt':
with open(file_path, 'rb') as f:
raw = f.read()
try:
content = raw.decode('utf-8')
except UnicodeDecodeError as exc:
raise ValueError("Invalid file content: TXT must be valid UTF-8 text") from exc
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@backend/Generator/main.py` around lines 406 - 412, The current TXT handling
only decodes the 512-byte header (header) which misses invalid UTF-8 later in
the file; change the logic for the ext == 'txt' branch to read the full file
bytes from file_path (open in 'rb'), attempt to decode the entire payload with
utf-8 and if a UnicodeDecodeError occurs raise ValueError("Invalid file content:
TXT must be valid UTF-8 text"); on successful decode assign the decoded string
to content (or use it when opening as text) so the full file is validated before
any further processing.

else:
raise ValueError("Unsupported file extension")

finally:
if os.path.exists(file_path):
os.remove(file_path)
Comment thread
coderabbitai[bot] marked this conversation as resolved.

os.remove(file_path)
return content

class QuestionGenerator:
Expand Down
19 changes: 13 additions & 6 deletions backend/server.py
Original file line number Diff line number Diff line change
Expand Up @@ -495,12 +495,19 @@ def upload_file():
if file.filename == '':
return jsonify({"error": "No selected file"}), 400

content = file_processor.process_file(file)

if content:
return jsonify({"content": content})
else:
return jsonify({"error": "Unsupported file type or error processing file"}), 400
try:
content = file_processor.process_file(file)

if content:
return jsonify({"content": content})
else:
return jsonify({"error": "Unsupported file type or error processing file"}), 400
except ValueError as e:
app.logger.warning("Upload validation failed: %s", e)
return jsonify({"error": str(e)}), 400
except Exception as e:
app.logger.exception("Error processing upload: %s", e)
return jsonify({"error": "Internal server error"}), 500

@app.route("/", methods=["GET"])
def hello():
Expand Down