Security Hardening

Solution

To achieve defense-in-depth, configure a multi-layered security model across your custom Frappe app. This includes parameterized queries, HTML sanitization, automated CSRF, request rate limiting, file upload validation, environment secrets management, and custom security middleware.

SQL Injection Prevention

To prevent SQL Injection (SQLi) in custom Frappe apps, never interpolate user input directly into SQL queries. Instead, utilize Frappe’s ORM or parameterized queries.

Here are five vulnerable patterns and their secure equivalents:

"""
SQL Injection Prevention Guide for Frappe Developers.

NEVER deploy the vulnerable patterns. They exist here only for education.
"""
import frappe


# === PATTERN 1: String interpolation in frappe.db.sql ===

# VULNERABLE -- attacker controls `outlet_name`
def get_outlet_vulnerable_1(outlet_name):
    return frappe.db.sql(
        f"SELECT * FROM `tabFranchise Outlet` WHERE outlet_name = '{outlet_name}'"
    )
    # Attack: outlet_name = "' OR 1=1 --"

# SAFE -- parameterized query
def get_outlet_safe_1(outlet_name):
    return frappe.db.sql(
        "SELECT * FROM `tabFranchise Outlet` WHERE outlet_name = %s",
        (outlet_name,),
        as_dict=True,
    )


# === PATTERN 2: f-string in ORDER BY ===

# VULNERABLE -- attacker controls `sort_field`
def list_outlets_vulnerable_2(sort_field):
    return frappe.db.sql(
        f"SELECT name, city FROM `tabFranchise Outlet` ORDER BY {sort_field}"
    )
    # Attack: sort_field = "1; DROP TABLE `tabFranchise Outlet`;--"

# SAFE -- whitelist allowed fields
ALLOWED_SORT_FIELDS = {"outlet_name", "city", "creation", "monthly_rent"}

def list_outlets_safe_2(sort_field):
    if sort_field not in ALLOWED_SORT_FIELDS:
        sort_field = "creation"
    return frappe.db.sql(
        f"SELECT name, city FROM `tabFranchise Outlet` ORDER BY `{sort_field}`",
        as_dict=True,
    )


# === PATTERN 3: Dynamic filters via .format() ===

# VULNERABLE -- attacker controls `status`
def filter_outlets_vulnerable_3(status):
    return frappe.db.sql(
        "SELECT name FROM `tabFranchise Outlet` WHERE status = '{}'".format(status)
    )

# SAFE -- use frappe.get_all with ORM filters
def filter_outlets_safe_3(status):
    return frappe.get_all(
        "Franchise Outlet",
        filters={"status": status},
        fields=["name", "outlet_name", "city"],
    )


# === PATTERN 4: LIKE clause with user input ===

# VULNERABLE -- unescaped wildcard injection
def search_outlets_vulnerable_4(search_term):
    return frappe.db.sql(
        f"SELECT name FROM `tabFranchise Outlet` WHERE city LIKE '%{search_term}%'"
    )

# SAFE -- parameterized LIKE
def search_outlets_safe_4(search_term):
    return frappe.db.sql(
        "SELECT name FROM `tabFranchise Outlet` WHERE city LIKE %s",
        (f"%{frappe.db.escape_like(search_term)}%",),
        as_dict=True,
    )


# === PATTERN 5: Dynamic table/field names from user input ===

# VULNERABLE -- attacker controls `doctype_name`
def count_records_vulnerable_5(doctype_name):
    return frappe.db.sql(f"SELECT COUNT(*) FROM `tab{doctype_name}`")[0][0]
    # Attack: doctype_name = "User` WHERE 1=1 UNION SELECT password FROM `__Auth"

# SAFE -- validate doctype exists and user has permission
def count_records_safe_5(doctype_name):
    if not frappe.db.exists("DocType", doctype_name):
        frappe.throw("Invalid DocType")
    frappe.has_permission(doctype_name, "read", throw=True)
    return frappe.db.count(doctype_name)

XSS Prevention

Cross-Site Scripting (XSS) occurs when untrusted HTML or JavaScript is rendered in the user’s browser. Mitigate XSS at the controller layer, Jinja template layer, and client script layer:

"""XSS Prevention patterns for Frappe apps."""
import frappe
from frappe.utils import sanitize_html, strip_html


# --- In Python controllers: sanitize before save ---
class FranchiseOutlet:
    def validate(self):
        # Sanitize free-text fields that render as HTML
        if self.outlet_description:
            self.outlet_description = sanitize_html(self.outlet_description)

        # Strip HTML entirely from fields that should be plain text
        if self.outlet_name:
            self.outlet_name = strip_html(self.outlet_name)

In Jinja print formats and web templates, variables rendered inside {{ }} are auto-escaped by default. For HTML fields (like rich text), use the |sanitize_html filter instead of the dangerous |safe filter.

<!-- JINJA TEMPLATE: XSS Prevention -->

<!-- VULNERABLE: raw output -->
<!-- <h3>{{ outlet.outlet_name }}</h3> -->

<!-- SAFE: Jinja auto-escapes by default with {{ }} -->
<h3>{{ outlet.outlet_name }}</h3>

<!-- When you MUST render HTML (e.g., rich text), use the |sanitize filter -->
<div class="description">{{ outlet.description | sanitize_html }}</div>

<!-- NEVER use |safe on user-supplied content -->
<!-- BAD: {{ outlet.description | safe }} -->

On the client side, use standard jQuery escaping methods (like .text()) or frappe.render_template instead of modifying innerHTML directly:

// Client Script: XSS Prevention

frappe.ui.form.on("Franchise Outlet", {
    refresh(frm) {
        // VULNERABLE: innerHTML with user data
        // document.getElementById("outlet-info").innerHTML = frm.doc.notes;

        // SAFE: Use frappe's built-in methods that escape HTML
        const $wrapper = frm.fields_dict.outlet_info_html.$wrapper;
        $wrapper.empty();
        $wrapper.append(
            $("<div>").text(frm.doc.notes)  // jQuery .text() escapes HTML
        );

        // SAFE: Using frappe.render_template (auto-escapes)
        const html = frappe.render_template(
            '<div class="info">{{ notes }}</div>',
            { notes: frm.doc.notes }
        );
        $wrapper.html(html);
    },
});

CSRF Protection

Cross-Site Request Forgery (CSRF) is handled automatically by Frappe’s Werkzeug-based server for all session-based authentication.

"""
Frappe's CSRF protection is automatic for session-based auth.

Every response includes a cookie: X-Frappe-CSRF-Token
Every POST/PUT/DELETE request must include this token as a header
or form field named 'csrf_token'.

For API key/token auth, CSRF is not required (no session cookie).
"""
import frappe


# Example: calling from JavaScript (CSRF handled automatically by frappe.call)
# frappe.call({
#     method: "scoopjoy.api.mobile.update_outlet_status",
#     args: { outlet: "SJ-BLR-001", status: "Inactive" },
# })
# frappe.call automatically includes X-Frappe-CSRF-Token header

# Example: calling from external client with token auth (no CSRF needed)
# curl -X POST https://erp.scoopjoy.com/api/method/scoopjoy.api.mobile.update_outlet_status \
#   -H "Authorization: token api_key:api_secret" \
#   -H "Content-Type: application/json" \
#   -d '{"outlet": "SJ-BLR-001", "status": "Inactive"}'

Rate Limiting Configuration

Protect endpoints against brute-force and resource-exhaustion attacks using global and per-method rate limits.

First, set up global rate limiting in sites/common_site_config.json:

{
    "rate_limit": {
        "limit": 600,
        "window": 3600
    }
}

Then, use the @rate_limit decorator on specific whitelisted Python API endpoints:

# Per-endpoint rate limiting using the decorator
import frappe
from frappe.rate_limiter import rate_limit


@frappe.whitelist()
@rate_limit(limit=10, seconds=60)  # 10 requests per minute
def get_item_availability(outlet):
    """Rate-limited endpoint for POS item lookups."""
    frappe.has_permission("Franchise Outlet", "read", outlet, throw=True)
    # ... business logic ...
    return {"items": []}


@frappe.whitelist(allow_guest=True)
@rate_limit(limit=5, seconds=300)  # 5 requests per 5 minutes -- strict for login
def custom_login(usr, pwd):
    """Custom login with aggressive rate limiting."""
    frappe.local.login_manager.authenticate(usr, pwd)
    frappe.local.login_manager.post_login()
    return {"message": "Logged in"}

Secure File Upload Handling

Prevent malicious file uploads by implementing strict extension, MIME type, size, and path traversal validation.

Implement validation logic: Validate uploaded files before insert using before_insert hook logic.

import frappe
import os
import mimetypes

# Allowed file types for franchise documents
ALLOWED_EXTENSIONS = {".pdf", ".jpg", ".jpeg", ".png", ".xlsx", ".csv"}
ALLOWED_MIMETYPES = {
    "application/pdf",
    "image/jpeg",
    "image/png",
    "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
    "text/csv",
}
MAX_FILE_SIZE_MB = 10


def validate_file_upload(doc, method):
    """Hook: before_insert on File doctype."""
    if not doc.file_name:
        return

    # 1. Extension validation
    _, ext = os.path.splitext(doc.file_name.lower())
    if ext not in ALLOWED_EXTENSIONS:
        frappe.throw(
            f"File type '{ext}' is not allowed. "
            f"Permitted: {', '.join(sorted(ALLOWED_EXTENSIONS))}",
            title="Invalid File Type",
        )

    # 2. MIME type validation (don't trust extension alone)
    if doc.content:
        mime_type = mimetypes.guess_type(doc.file_name)[0]
        if mime_type and mime_type not in ALLOWED_MIMETYPES:
            frappe.throw(
                f"MIME type '{mime_type}' is not allowed.",
                title="Invalid File Type",
            )

    # 3. Size validation
    if doc.file_size and doc.file_size > MAX_FILE_SIZE_MB * 1024 * 1024:
        frappe.throw(
            f"File size exceeds {MAX_FILE_SIZE_MB}MB limit.",
            title="File Too Large",
        )

    # 4. Path traversal prevention
    if ".." in doc.file_name or "/" in doc.file_name or "\\" in doc.file_name:
        frappe.throw("Invalid file name.", title="Security Error")

    # 5. Sanitize filename
    doc.file_name = frappe.utils.escape_html(
        os.path.basename(doc.file_name)
    )

Register the DocType event hook: Add the validation function under the File document events in hooks.py.

doc_events = {
    "File": {
        "before_insert": "scoopjoy.overrides.file_upload.validate_file_upload",
    },
}

API Key Rotation Pattern

API keys should be rotated periodically. To do this securely, generate new credentials using cryptographically secure hashes and immediately update the user document.

import frappe
from frappe.utils import add_days, now_datetime, getdate


@frappe.whitelist()
def rotate_api_key():
    """Rotate the current user's API key. Old key invalidated immediately."""
    user = frappe.session.user
    if user == "Administrator" or user == "Guest":
        frappe.throw("Cannot rotate API key for this user.")

    user_doc = frappe.get_doc("User", user)

    # Generate new credentials
    new_api_key = frappe.generate_hash(length=15)
    new_api_secret = frappe.generate_hash(length=15)

    user_doc.api_key = new_api_key
    user_doc.api_secret = new_api_secret
    user_doc.save(ignore_permissions=True)

    # Log the rotation event
    frappe.get_doc(
        {
            "doctype": "Activity Log",
            "subject": f"API key rotated for {user}",
            "user": user,
            "operation": "API Key Rotation",
        }
    ).insert(ignore_permissions=True)

    return {
        "api_key": new_api_key,
        "api_secret": new_api_secret,
        "message": "API key rotated. Update your client applications.",
    }

Audit Logging for Sensitive Operations

Implement explicit logging for key operations by inserting records into the Activity Log DocType.

import frappe
import json


def log_sensitive_operation(doctype, docname, operation, details=None):
    """Create an audit log entry for sensitive operations."""
    frappe.get_doc(
        {
            "doctype": "Activity Log",
            "subject": f"{operation} on {doctype} {docname}",
            "content": json.dumps(details or {}, indent=2, default=str),
            "user": frappe.session.user,
            "operation": operation,
            "reference_doctype": doctype,
            "reference_name": docname,
            "ip_address": frappe.local.request_ip if frappe.local.request else None,
        }
    ).insert(ignore_permissions=True)


# Usage in controller:
class FranchiseAgreement:
    def on_submit(self):
        log_sensitive_operation(
            self.doctype, self.name, "Agreement Submitted",
            details={
                "franchise_outlet": self.franchise_outlet,
                "royalty_percentage": self.royalty_percentage,
                "security_deposit": self.security_deposit,
            },
        )

    def on_cancel(self):
        log_sensitive_operation(
            self.doctype, self.name, "Agreement Cancelled",
            details={"cancelled_by": frappe.session.user},
        )

Environment Variable Management

To avoid exposing credentials, secrets should never be hardcoded. Store them in site_config.json or fallback to OS environment variables.

"""Never hardcode secrets. Use site_config.json or environment variables."""
import os
import frappe


def get_secret(key, required=True):
    """Retrieve a secret from site_config or environment variable.

    Priority: site_config.json > environment variable
    """
    # Try site_config first
    value = frappe.conf.get(key)

    # Fall back to environment variable
    if not value:
        env_key = f"SCOOPJOY_{key.upper()}"
        value = os.environ.get(env_key)

    if required and not value:
        frappe.throw(
            f"Missing required secret: {key}. "
            f"Set it in site_config.json or as env var SCOOPJOY_{key.upper()}",
            title="Configuration Error",
        )

    return value


# Usage:
# payment_gateway_key = get_secret("razorpay_key_id")
# sms_api_key = get_secret("twilio_auth_token")

# In site_config.json:
# {
#     "razorpay_key_id": "rzp_live_xxxx",
#     "razorpay_key_secret": "yyyy",
#     "twilio_auth_token": "zzzz"
# }

Complete Security Middleware

Enforce HTTPS, secure headers, and content types before the request handler processes the payload.

Define the middleware functions: Implement the checks in a middleware module.

"""Security middleware applied via hooks.py app_include_* or before_request."""
import frappe
import re


def before_request():
    """Security checks run before every request."""
    _enforce_https_redirect()
    _set_security_headers()
    _validate_content_type()
    _log_suspicious_requests()


def _enforce_https_redirect():
    """Redirect HTTP to HTTPS in production."""
    if not frappe.local.conf.get("developer_mode"):
        if frappe.local.request and not frappe.local.request.is_secure:
            secure_url = frappe.local.request.url.replace("http://", "https://", 1)
            frappe.local.response["type"] = "redirect"
            frappe.local.response["location"] = secure_url


def _set_security_headers():
    """Add security headers to every response."""
    headers = frappe.local.response.headers if hasattr(frappe.local.response, "headers") else {}
    security_headers = {
        "X-Content-Type-Options": "nosniff",
        "X-Frame-Options": "SAMEORIGIN",
        "X-XSS-Protection": "1; mode=block",
        "Referrer-Policy": "strict-origin-when-cross-origin",
        "Permissions-Policy": "camera=(), microphone=(), geolocation=()",
        "Strict-Transport-Security": "max-age=31536000; includeSubDomains",
    }
    for key, value in security_headers.items():
        if hasattr(frappe.local, "_response"):
            frappe.local._response.headers[key] = value


# Suspicious patterns to flag
SUSPICIOUS_PATTERNS = [
    re.compile(r"(\bunion\b.*\bselect\b)", re.IGNORECASE),
    re.compile(r"(<script|javascript:|on\w+=)", re.IGNORECASE),
    re.compile(r"(\.\./|\.\.\\)", re.IGNORECASE),  # path traversal
    re.compile(r"(;\s*drop\s+table)", re.IGNORECASE),
]


def _validate_content_type():
    """Reject requests with unexpected content types for POST."""
    if frappe.local.request and frappe.local.request.method == "POST":
        content_type = frappe.local.request.content_type or ""
        allowed = {
            "application/x-www-form-urlencoded",
            "multipart/form-data",
            "application/json",
        }
        base_type = content_type.split(";")[0].strip().lower()
        if base_type and base_type not in allowed:
            frappe.throw("Unsupported Content-Type", frappe.AuthenticationError)


def _log_suspicious_requests():
    """Log requests containing potential attack patterns."""
    if not frappe.local.request:
        return

    request_data = str(frappe.local.request.url) + str(frappe.local.form_dict)
    for pattern in SUSPICIOUS_PATTERNS:
        if pattern.search(request_data):
            frappe.logger("security").warning(
                f"Suspicious request from {frappe.local.request_ip}: "
                f"pattern={pattern.pattern}, url={frappe.local.request.url}, "
                f"user={frappe.session.user}"
            )
            break

Register the request middleware: Bind the security checks under before_request in hooks.py.
apps/scoopjoy/scoopjoy/hooks.py
```
before_request = ["scoopjoy.middleware.before_request"]
```