Fix SMTP defaults and upload validation

This commit is contained in:
A R R R Associates
2026-06-28 09:10:49 +05:30
parent f5174575c5
commit df4fd17211
7 changed files with 220 additions and 26 deletions
+72 -2
View File
@@ -10,6 +10,7 @@ from sqlalchemy.orm import Session, joinedload
from app.core.settings import get_settings
from app.modules.clients.models import Client
from app.modules.core.tenancy.settings_models import BranchSettings
from app.modules.core.rbac.deps import get_user_permissions, get_user_roles
from app.modules.notice_cases.models import (
NoticeCase,
@@ -381,6 +382,75 @@ def add_order(db: Session, *, case: NoticeCase, user, data: dict) -> NoticeCaseO
return row
class UploadValidationError(ValueError):
"""Raised when a notice/case document upload violates branch file policy."""
DANGEROUS_DOCUMENT_EXTENSIONS = {
".exe", ".bat", ".cmd", ".com", ".scr", ".pif",
".sh", ".bash", ".zsh", ".ps1", ".psm1",
".js", ".jse", ".vbs", ".vbe", ".wsf",
".msi", ".dll", ".jar", ".apk",
}
def _branch_document_settings(db: Session, branch_id: int | None) -> BranchSettings | None:
if not branch_id:
return None
return db.execute(
select(BranchSettings).where(BranchSettings.branch_id == int(branch_id))
).scalar_one_or_none()
def _normalise_allowed_extensions(raw: str | None) -> set[str]:
values = raw or "pdf,jpg,jpeg,png,xlsx,xls,docx,zip"
return {
"." + item.strip().lower().lstrip(".")
for item in values.split(",")
if item.strip()
}
def notice_case_upload_policy(db: Session, *, branch_id: int | None) -> dict[str, object]:
settings = _branch_document_settings(db, branch_id)
allowed_exts = _normalise_allowed_extensions(getattr(settings, "allowed_ext_csv", None))
max_file_mb = int(getattr(settings, "max_file_mb", 25) or 25)
if max_file_mb <= 0:
max_file_mb = 25
return {
"allowed_exts": sorted(allowed_exts),
"allowed_ext_csv": ",".join(ext.lstrip(".") for ext in sorted(allowed_exts)),
"accept": ",".join(sorted(allowed_exts)),
"max_file_mb": max_file_mb,
"max_file_bytes": max_file_mb * 1024 * 1024,
}
def _validate_notice_case_upload_policy(db: Session, *, case: NoticeCase, upload_file: UploadFile, original: str) -> bytes:
policy = notice_case_upload_policy(db, branch_id=case.branch_id)
allowed_exts = set(policy["allowed_exts"] or [])
max_file_bytes = int(policy["max_file_bytes"] or (25 * 1024 * 1024))
if not original or original in {".", ".."}:
raise UploadValidationError("Invalid filename.")
ext = Path(original).suffix.lower()
if not ext:
raise UploadValidationError("File extension is required.")
if ext in DANGEROUS_DOCUMENT_EXTENSIONS:
raise UploadValidationError("Executable or dangerous file types are not allowed.")
if allowed_exts and ext not in allowed_exts:
raise UploadValidationError("File type not allowed. Allowed types: " + ", ".join(sorted(allowed_exts)))
data = upload_file.file.read(max_file_bytes + 1)
if len(data) > max_file_bytes:
raise UploadValidationError(f"File too large. Maximum allowed size is {policy['max_file_mb']} MB.")
return data
def _storage_root() -> Path:
settings = get_settings()
base = getattr(settings, "LOCAL_STORAGE_ROOT", None) or getattr(settings, "DOCUMENT_STORAGE_ROOT", None) or "data/storage"
@@ -392,14 +462,14 @@ def case_document_absolute_path(document: NoticeCaseDocument) -> Path:
def save_case_document(db: Session, *, case: NoticeCase, upload_file: UploadFile, user, title: str, document_type: str, description: str | None, event_id: int | None = None) -> NoticeCaseDocument:
original = Path(upload_file.filename or "case_document.bin").name
original = Path(upload_file.filename or "").name
data = _validate_notice_case_upload_policy(db, case=case, upload_file=upload_file, original=original)
ext = Path(original).suffix.lower()
stored = f"case_{case.id}_{uuid4().hex}{ext}"
fy_folder = f"FY{case.financial_year}" if case.financial_year else "FY_UNASSIGNED"
rel = Path("notice_cases") / fy_folder / str(case.tenant_id) / str(case.client_id) / case.case_code / stored
absolute = _storage_root() / rel
absolute.parent.mkdir(parents=True, exist_ok=True)
data = upload_file.file.read()
absolute.write_bytes(data)
latest_version = db.execute(select(func.max(NoticeCaseDocument.version_no)).where(NoticeCaseDocument.case_id == case.id, NoticeCaseDocument.title == (title or original))).scalar_one() or 0
row = NoticeCaseDocument(
@@ -27,6 +27,6 @@
<section class="rounded-xl bg-white p-5 shadow-sm ring-1 ring-slate-200"><h2 class="font-bold">Orders</h2><form method="post" action="/notice-cases/{{ case.id }}/orders" class="mt-4 grid gap-3 md:grid-cols-2"><input type="hidden" name="csrf_token" value="{{ csrf_token }}"><select name="order_type" class="rounded-lg border px-3 py-2 text-sm">{% for o in order_types %}<option>{{ o }}</option>{% endfor %}</select><input name="order_no" placeholder="Order No" class="rounded-lg border px-3 py-2 text-sm"><input type="date" name="order_date" required class="rounded-lg border px-3 py-2 text-sm"><input type="date" name="appeal_due_date" class="rounded-lg border px-3 py-2 text-sm"><input name="tax_amount" placeholder="Tax" type="number" class="rounded-lg border px-3 py-2 text-sm"><input name="interest_amount" placeholder="Interest" type="number" class="rounded-lg border px-3 py-2 text-sm"><input name="penalty_amount" placeholder="Penalty" type="number" class="rounded-lg border px-3 py-2 text-sm"><input name="demand_amount" placeholder="Total Demand" type="number" class="rounded-lg border px-3 py-2 text-sm"><textarea name="summary" placeholder="Order summary" class="rounded-lg border px-3 py-2 text-sm md:col-span-2"></textarea><label class="text-sm md:col-span-2"><input type="checkbox" name="appeal_filed" value="1"> Appeal filed</label><button class="rounded-lg bg-slate-900 px-4 py-2 text-sm font-semibold text-white md:col-span-2">Add Order</button></form><div class="mt-4 space-y-2 text-sm">{% for o in case.orders %}<div class="rounded-lg bg-slate-50 p-3"><b>{{ o.order_type }}</b> · {{ o.order_date }} · Demand: {{ o.demand_amount }}<br>{{ o.summary or '' }}</div>{% endfor %}</div></section>
</div>
<section class="rounded-xl bg-white p-5 shadow-sm ring-1 ring-slate-200"><div class="flex items-center justify-between"><h2 class="font-bold">Case Documents</h2></div>{% if can_upload %}<form method="post" enctype="multipart/form-data" action="/notice-cases/{{ case.id }}/documents/upload" class="mt-4 grid gap-3 md:grid-cols-5"><input type="hidden" name="csrf_token" value="{{ csrf_token }}"><input name="title" placeholder="Title" class="rounded-lg border px-3 py-2 text-sm"><select name="document_type" class="rounded-lg border px-3 py-2 text-sm">{% for d in case_document_types %}<option value="{{ d }}">{{ d }}</option>{% endfor %}</select><input name="description" placeholder="Description" class="rounded-lg border px-3 py-2 text-sm"><input type="file" name="file" required class="rounded-lg border px-3 py-2 text-sm"><button class="rounded-lg bg-brand-600 px-4 py-2 text-sm font-semibold text-white">Upload</button></form>{% endif %}<div class="mt-4 overflow-hidden rounded-lg border"><table class="min-w-full divide-y divide-slate-200 text-sm"><thead class="bg-slate-50"><tr><th class="px-3 py-2 text-left">Document</th><th class="px-3 py-2">Type</th><th class="px-3 py-2">Uploaded</th><th class="px-3 py-2 text-right">Action</th></tr></thead><tbody class="divide-y divide-slate-100">{% for d in case.documents if not d.is_deleted %}<tr><td class="px-3 py-2"><div class="font-semibold">{{ d.title }}</div><div class="text-xs text-slate-500">{{ d.original_filename }}</div></td><td class="px-3 py-2 text-center">{{ d.document_type }}</td><td class="px-3 py-2 text-center">{{ d.uploaded_at_utc.date() if d.uploaded_at_utc else '' }}</td><td class="px-3 py-2 text-right"><a href="/notice-cases/documents/{{ d.id }}/download" class="font-semibold text-brand-700">Download</a>{% if can_delete_documents %}<form method="post" action="/notice-cases/documents/{{ d.id }}/delete" class="inline"><input type="hidden" name="csrf_token" value="{{ csrf_token }}"><button class="ml-3 text-red-600" onclick="return confirm('Delete this document?')">Delete</button></form>{% endif %}</td></tr>{% else %}<tr><td colspan="4" class="px-3 py-6 text-center text-slate-500">No documents uploaded.</td></tr>{% endfor %}</tbody></table></div></section>
<section class="rounded-xl bg-white p-5 shadow-sm ring-1 ring-slate-200"><div class="flex items-center justify-between"><h2 class="font-bold">Case Documents</h2></div>{% if upload_error %}<div class="mt-4 rounded-lg border border-red-200 bg-red-50 px-4 py-3 text-sm font-medium text-red-700">{{ upload_error }}</div>{% endif %}{% if can_upload %}<form method="post" enctype="multipart/form-data" action="/notice-cases/{{ case.id }}/documents/upload" class="mt-4 grid gap-3 md:grid-cols-5"><input type="hidden" name="csrf_token" value="{{ csrf_token }}"><input name="title" placeholder="Title" class="rounded-lg border px-3 py-2 text-sm"><select name="document_type" class="rounded-lg border px-3 py-2 text-sm">{% for d in case_document_types %}<option value="{{ d }}">{{ d }}</option>{% endfor %}</select><input name="description" placeholder="Description" class="rounded-lg border px-3 py-2 text-sm"><input type="file" name="file" required accept="{{ upload_policy.accept if upload_policy else '.pdf,.jpg,.jpeg,.png,.xlsx,.xls,.docx,.zip' }}" class="rounded-lg border px-3 py-2 text-sm"><button class="rounded-lg bg-brand-600 px-4 py-2 text-sm font-semibold text-white">Upload</button><p class="text-xs text-slate-500 md:col-span-5">Allowed file types: {{ upload_policy.allowed_ext_csv if upload_policy else 'pdf,jpg,jpeg,png,xlsx,xls,docx,zip' }}. Maximum size: {{ upload_policy.max_file_mb if upload_policy else 25 }} MB.</p></form>{% endif %}<div class="mt-4 overflow-hidden rounded-lg border"><table class="min-w-full divide-y divide-slate-200 text-sm"><thead class="bg-slate-50"><tr><th class="px-3 py-2 text-left">Document</th><th class="px-3 py-2">Type</th><th class="px-3 py-2">Uploaded</th><th class="px-3 py-2 text-right">Action</th></tr></thead><tbody class="divide-y divide-slate-100">{% for d in case.documents if not d.is_deleted %}<tr><td class="px-3 py-2"><div class="font-semibold">{{ d.title }}</div><div class="text-xs text-slate-500">{{ d.original_filename }}</div></td><td class="px-3 py-2 text-center">{{ d.document_type }}</td><td class="px-3 py-2 text-center">{{ d.uploaded_at_utc.date() if d.uploaded_at_utc else '' }}</td><td class="px-3 py-2 text-right"><a href="/notice-cases/documents/{{ d.id }}/download" class="font-semibold text-brand-700">Download</a>{% if can_delete_documents %}<form method="post" action="/notice-cases/documents/{{ d.id }}/delete" class="inline"><input type="hidden" name="csrf_token" value="{{ csrf_token }}"><button class="ml-3 text-red-600" onclick="return confirm('Delete this document?')">Delete</button></form>{% endif %}</td></tr>{% else %}<tr><td colspan="4" class="px-3 py-6 text-center text-slate-500">No documents uploaded.</td></tr>{% endfor %}</tbody></table></div></section>
</div>
{% endblock %}
+36 -4
View File
@@ -36,6 +36,8 @@ from app.modules.notice_cases.service import (
list_clients_for_case,
list_engagements_for_client,
save_case_document,
notice_case_upload_policy,
UploadValidationError,
update_case,
user_can_access_case,
)
@@ -72,8 +74,8 @@ def _base_ctx(request: Request, user, db, **ctx):
return base
def _render(request: Request, template_name: str, db, user, **ctx):
return templates.TemplateResponse(template_name, _base_ctx(request, user, db, **ctx))
def _render(request: Request, template_name: str, db, user, status_code: int = 200, **ctx):
return templates.TemplateResponse(template_name, _base_ctx(request, user, db, **ctx), status_code=status_code)
def _redirect_denied():
@@ -298,7 +300,19 @@ def case_detail(request: Request, case_id: int):
case = get_case(db, case_id)
if not case or not _case_access_allowed(request, db, user, case):
return _redirect_denied()
return _render(request, "modules/notice_cases/templates/notice_cases/detail.html", db, user, title=case.case_code, case=case, can_manage=can_manage_notice_cases(db, user), can_upload=can_upload_notice_case_documents(db, user), can_delete_documents=can_delete_notice_case_documents(db, user))
upload_policy = notice_case_upload_policy(db, branch_id=case.branch_id)
return _render(
request,
"modules/notice_cases/templates/notice_cases/detail.html",
db,
user,
title=case.case_code,
case=case,
can_manage=can_manage_notice_cases(db, user),
can_upload=can_upload_notice_case_documents(db, user),
can_delete_documents=can_delete_notice_case_documents(db, user),
upload_policy=upload_policy,
)
finally:
db.close()
@@ -445,7 +459,25 @@ def case_document_upload(request: Request, case_id: int, title: str = Form(""),
return RedirectResponse(url=f"/notice-cases/{case.id}?year_locked=1", status_code=303)
if not file or not file.filename:
return RedirectResponse(url=f"/notice-cases/{case_id}?error=missing_file", status_code=303)
save_case_document(db, case=case, upload_file=file, user=user, title=title, document_type=document_type, description=description, event_id=int(event_id) if event_id else None)
try:
save_case_document(db, case=case, upload_file=file, user=user, title=title, document_type=document_type, description=description, event_id=int(event_id) if event_id else None)
except UploadValidationError as exc:
db.rollback()
upload_policy = notice_case_upload_policy(db, branch_id=case.branch_id)
return _render(
request,
"modules/notice_cases/templates/notice_cases/detail.html",
db,
user,
status_code=400,
title=case.case_code,
case=case,
can_manage=can_manage_notice_cases(db, user),
can_upload=can_upload_notice_case_documents(db, user),
can_delete_documents=can_delete_notice_case_documents(db, user),
upload_policy=upload_policy,
upload_error=str(exc),
)
db.commit()
return RedirectResponse(url=f"/notice-cases/{case_id}?uploaded=1", status_code=303)
finally: