Is Your Startup Ready for DPDPA? A Founder's Reality Check

<!-- WRONG: Vague consent -->
<input type="checkbox" checked>
  <label>I agree to terms and conditions</label>

<!-- RIGHT: Specific, granular consent -->
<fieldset>
  <legend>Data Collection Consent</legend>
  <div>
    <input type="checkbox" name="consent_marketing" id="consent_marketing">
    <label for="consent_marketing">
      I consent to receive marketing emails (Purpose: Email campaigns)
    </label>
  </div>
  <div>
    <input type="checkbox" name="consent_analytics" id="consent_analytics">
    <label for="consent_analytics">
      I consent to analytics tracking (Purpose: Product improvement)
    </label>
  </div>
  <div>
    <input type="checkbox" name="consent_third_party" id="consent_third_party">
    <label for="consent_third_party">
      I consent to share data with partners (Partners: [list])
    </label>
  </div>
</fieldset>

#!/bin/bash
# Quick data inventory audit for your infrastructure

# Find unencrypted databases
echo "=== Checking for unencrypted databases ==="
grep -r "mongodb://" . --include="*.env" --include="*.config"
grep -r "postgresql://" . --include="*.env" --include="*.config"

# Find hardcoded credentials
echo "=== Scanning for exposed secrets ==="
grep -r "password\|api_key\|secret" . --include="*.js" --include="*.py" | head -20

# Check file permissions on sensitive data
echo "=== Checking file permissions ==="
find . -name "*.csv" -o -name "*.xlsx" -o -name "*.json" | xargs ls -la | grep -v "^d"

# Identify cloud storage buckets
echo "=== AWS S3 buckets with public access ==="
aws s3api list-buckets --query 'Buckets[].Name' | xargs -I {} aws s3api get-bucket-acl --bucket {} --query 'Grants[?Grantee.Type==`Group`]'

# Example: Audit logging for data access
import logging
import json
from datetime import datetime

class DataAccessAuditLogger:
    def __init__(self, log_file="data_access_audit.log"):
        self.logger = logging.getLogger("DataAccess")
        handler = logging.FileHandler(log_file)
        formatter = logging.Formatter(
            '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
        )
        handler.setFormatter(formatter)
        self.logger.addHandler(handler)
        self.logger.setLevel(logging.INFO)
    
    def log_access(self, user_id, data_type, action, ip_address, reason):
        """Log when personal data is accessed"""
        audit_entry = {
            "timestamp": datetime.utcnow().isoformat(),
            "user_id": user_id,
            "data_type": data_type,  # e.g., "customer_email", "payment_info"
            "action": action,  # e.g., "read", "delete", "export"
            "ip_address": ip_address,
            "reason": reason  # e.g., "customer_support_request"
        }
        self.logger.info(json.dumps(audit_entry))
    
    def log_breach(self, affected_records, breach_type, severity):
        """Log suspected data breach"""
        breach_entry = {
            "timestamp": datetime.utcnow().isoformat(),
            "affected_records": affected_records,
            "breach_type": breach_type,  # e.g., "unauthorized_access", "sql_injection"
            "severity": severity,  # e.g., "critical", "high"
            "action_taken": "CERT-In notification initiated"
        }
        self.logger.error(json.dumps(breach_entry))

# Usage
audit = DataAccessAuditLogger()
audit.log_access(
    user_id="emp_12345",
    data_type="customer_email",
    action="read",
    ip_address="192.168.1.100",
    reason="customer_support_ticket_#5678"
)

#!/bin/bash
# DPDP Breach Response Checklist

echo "=== IMMEDIATE (Within 1 Hour) ==="
echo "[ ] Isolate affected systems"
echo "[ ] Preserve evidence/logs"
echo "[ ] Assemble incident response team"
echo "[ ] Document timeline of discovery"

echo ""
echo "=== CRITICAL (Within 6 Hours - CERT-In) ==="
echo "[ ] Submit incident report to CERT-In"
echo "[ ] Include: affected data types, volume, breach vector, impact"
echo "[ ] Contact details for follow-up"
echo "[ ] Provide evidence of containment"

echo ""
echo "=== HIGH (Within 72 Hours - Users) ==="
echo "[ ] Notify all affected data subjects"
echo "[ ] Include: what data was compromised, breach date, your response"
echo "[ ] Provide: credit monitoring (if financial data), support contact"
echo "[ ] Send via: registered email, SMS, or secure portal"

echo ""
echo "=== FOLLOW-UP (Within 30 Days) ==="
echo "[ ] Complete forensic investigation"
echo "[ ] File formal report with Data Protection Board"
echo "[ ] Implement remediation measures"
echo "[ ] Communicate fixes to affected users"

# PostgreSQL: Enable encryption at rest
# 1. Generate encryption key
openssl rand -hex 32 > /etc/postgresql/encryption.key
chmod 600 /etc/postgresql/encryption.key

# 2. Enable pgcrypto extension
sudo -u postgres psql -c "CREATE EXTENSION pgcrypto;"

# 3. Encrypt sensitive columns
alter table customers add column email_encrypted bytea;
update customers 
set email_encrypted = pgp_sym_encrypt(email, 'your-encryption-key')
where email is not null;

# MySQL: Enable encryption at rest (InnoDB)
# Add to my.cnf
[mysqld]
default-table-encryption=ON
table-encryption-privilege-check=ON

# Restart MySQL
sudo systemctl restart mysql

# Flask example: Data subject rights API
from flask import Flask, request, jsonify
from datetime import datetime, timedelta
import json

app = Flask(__name__)

@app.route('/api/data-subject/access', methods=['POST'])
def request_data_access():
    """
    User requests their personal data (Right to Access)
    """
    data = request.json
    user_email = data.get('email')
    
    # Validate user identity
    if not validate_user(user_email):
        return jsonify({"error": "User not found"}), 404
    
    # Create request ticket
    request_id = generate_request_id()
    due_date = datetime.now() + timedelta(days=30)
    
    # Log the request
    log_data_subject_request(
        request_type="access",
        user_email=user_email,
        request_id=request_id,
        due_date=due_date
    )
    
    # Queue for fulfillment
    queue_fulfillment(request_id, user_email, "access")
    
    return jsonify({
        "request_id": request_id,
        "status": "pending",
        "due_date": due_date.isoformat(),
        "message": "Your data access request has been received. We'll respond within 30 days."
    }), 202

@app.route('/api/data-subject/erasure', methods=['POST'])
def request_data_erasure():
    """
    User requests deletion of their personal data (Right to Erasure)
    """
    data = request.json
    user_email = data.get('email')
    reason = data.get('reason', 'Not specified')
    
    # Validate user identity
    if not validate_user(user_email):
        return jsonify({"error": "User not found"}), 404
    
    # Check for legal holds
    if has_legal_hold(user_email):
        return jsonify({
            "error": "Erasure request cannot be fulfilled due to legal hold"
        }), 403
    
    # Create erasure request
    request_id = generate_request_id()
    due_date = datetime.now() + timedelta(days=30)
    
    log_data_subject_request(
        request_type="erasure",
        user_email=user_email,
        request_id=request_id,
        reason=reason,
        due_date=due_date
    )
    
    # Queue for fulfillment
    queue_fulfillment(request_id, user_email, "erasure")
    
    return jsonify({
        "request_id": request_id,
        "status": "pending",
        "due_date": due_date.isoformat(),
        "message": "Your data erasure request has been received. We'll process it within 30 days."
    }), 202

if __name__ == '__main__':
    app.run(ssl_context='adhoc')