pramoddutta/qaskills/owasp-security

import { test, expect } from '@playwright/test';
test.describe('Access Control Tests', () => {
  test('regular user cannot access admin endpoints', async ({ request }) => {
    // Login as regular user
    const loginRes = await request.post('/api/auth/login', {
      data: { email: 'user@example.com', password: 'UserPass123!' },
    });
    const { token } = await loginRes.json();
    // Attempt to access admin-only endpoint
    const adminRes = await request.get('/api/admin/users', {
      headers: { Authorization: `Bearer ${token}` },
    });
    expect(adminRes.status()).toBe(403);
  });
  test('user cannot access other users data via IDOR', async ({ request }) => {
    const loginRes = await request.post('/api/auth/login', {
      data: { email: 'user1@example.com', password: 'UserPass123!' },
    });
    const { token } = await loginRes.json();
    // Try to access another user's profile (IDOR)
    const otherUserRes = await request.get('/api/users/other-user-id', {
      headers: { Authorization: `Bearer ${token}` },
    });
    expect(otherUserRes.status()).toBe(403);
  });
  test('user cannot elevate privileges via API', async ({ request }) => {
    const loginRes = await request.post('/api/auth/login', {
      data: { email: 'user@example.com', password: 'UserPass123!' },
    });
    const { token } = await loginRes.json();
    // Try to change own role to admin
    const updateRes = await request.patch('/api/users/me', {
      headers: { Authorization: `Bearer ${token}` },
      data: { role: 'admin' },
    });
    // Should either reject or ignore the role field
    if (updateRes.status() === 200) {
      const body = await updateRes.json();
      expect(body.role).not.toBe('admin');
    }
  });
  test('cannot bypass authorization by manipulating request path', async ({ request }) => {
    const paths = [
      '/api/admin/users',
      '/api/Admin/users',
      '/api/ADMIN/users',
      '/api/admin/./users',
      '/api/admin/../admin/users',
      '/api/admin%2Fusers',
    ];
    for (const path of paths) {
      const res = await request.get(path);
      expect(res.status()).not.toBe(200);
    }
  });
});

test.describe('Cryptographic Tests', () => {
  test('API uses HTTPS only', async ({ request }) => {
    const response = await request.get('/', {
      headers: { Accept: 'text/html' },
    });
    const url = response.url();
    expect(url).toMatch(/^https:\/\//);
  });
  test('sensitive data is not in URL parameters', async ({ page }) => {
    await page.goto('/login');
    await page.getByLabel('Email').fill('user@example.com');
    await page.getByLabel('Password').fill('SecurePass123!');
    await page.getByRole('button', { name: 'Sign in' }).click();
    // Password should never appear in URL
    const url = page.url();
    expect(url).not.toContain('password');
    expect(url).not.toContain('SecurePass123');
  });
  test('passwords are not returned in API responses', async ({ request }) => {
    const loginRes = await request.post('/api/auth/login', {
      data: { email: 'user@example.com', password: 'UserPass123!' },
    });
    const body = await loginRes.json();
    const bodyStr = JSON.stringify(body);
    expect(bodyStr).not.toContain('UserPass123!');
    expect(body.user?.password).toBeUndefined();
    expect(body.user?.passwordHash).toBeUndefined();
  });
  test('cookies have Secure flag', async ({ page }) => {
    await page.goto('/');
    const cookies = await page.context().cookies();
    for (const cookie of cookies) {
      if (cookie.name.includes('session') || cookie.name.includes('token')) {
        expect(cookie.secure).toBe(true);
      }
    }
  });
});

test.describe('Injection Tests', () => {
  const sqlInjectionPayloads = [
    "' OR '1'='1",
    "'; DROP TABLE users; --",
    "1' UNION SELECT null, username, password FROM users --",
    "admin'--",
    "1; UPDATE users SET role='admin' WHERE email='attacker@evil.com'",
  ];
  const xssPayloads = [
    '<script>alert("XSS")</script>',
    '<img src=x onerror=alert("XSS")>',
    '"><script>alert("XSS")</script>',
    "javascript:alert('XSS')",
    '<svg onload=alert("XSS")>',
    '{{constructor.constructor("return this")()}}',
  ];
  for (const payload of sqlInjectionPayloads) {
    test(`SQL injection: ${payload.substring(0, 30)}...`, async ({ request }) => {
      const response = await request.get('/api/users/search', {
        params: { q: payload },
      });
      // Should not return 500 (indicates unhandled SQL error)
      expect(response.status()).not.toBe(500);
      const body = await response.text();
      // Should not contain SQL error messages
      expect(body).not.toContain('SQL syntax');
      expect(body).not.toContain('mysql_');
      expect(body).not.toContain('ORA-');
      expect(body).not.toContain('PostgreSQL');
    });
  }
  for (const payload of xssPayloads) {
    test(`XSS: ${payload.substring(0, 30)}...`, async ({ request }) => {
      // Submit XSS payload
      await request.post('/api/comments', {
        data: { content: payload, postId: 'test-post' },
      });
      // Retrieve and check the content is escaped
      const response = await request.get('/api/comments?postId=test-post');
      const body = await response.text();
      expect(body).not.toContain('<script>');
      expect(body).not.toContain('onerror=');
      expect(body).not.toContain('javascript:');
    });
  }
  test('NoSQL injection protection', async ({ request }) => {
    const response = await request.post('/api/auth/login', {
      data: {
        email: { $gt: '' },
        password: { $gt: '' },
      },
    });
    expect(response.status()).not.toBe(200);
  });
  test('Command injection protection', async ({ request }) => {
    const payloads = [
      '; ls -la',
      '| cat /etc/passwd',
      '`whoami`',
      '$(cat /etc/passwd)',
    ];
    for (const payload of payloads) {
      const response = await request.get('/api/ping', {
        params: { host: `example.com${payload}` },
      });
      const body = await response.text();
      expect(body).not.toContain('root:');
      expect(body).not.toContain('/bin/bash');
    }
  });
});

test.describe('Insecure Design Tests', () => {
  test('rate limiting on login endpoint', async ({ request }) => {
    const attempts = [];
    for (let i = 0; i < 20; i++) {
      attempts.push(
        request.post('/api/auth/login', {
          data: { email: 'user@example.com', password: `wrong${i}` },
        })
      );
    }
    const responses = await Promise.all(attempts);
    const rateLimited = responses.some((r) => r.status() === 429);
    expect(rateLimited).toBe(true);
  });
  test('account lockout after failed attempts', async ({ request }) => {
    // Attempt login 10 times with wrong password
    for (let i = 0; i < 10; i++) {
      await request.post('/api/auth/login', {
        data: { email: 'locktest@example.com', password: `wrong${i}` },
      });
    }
    // Even with correct password, should be locked
    const response = await request.post('/api/auth/login', {
      data: { email: 'locktest@example.com', password: 'CorrectPass123!' },
    });
    expect(response.status()).toBe(423); // Locked
  });
});

test.describe('Security Headers', () => {
  test('should have required security headers', async ({ request }) => {
    const response = await request.get('/');
    const headers = response.headers();
    // Content Security Policy
    expect(headers['content-security-policy']).toBeDefined();
    // Prevent MIME type sniffing
    expect(headers['x-content-type-options']).toBe('nosniff');
    // Prevent clickjacking
    expect(headers['x-frame-options']).toMatch(/DENY|SAMEORIGIN/);
    // HSTS
    expect(headers['strict-transport-security']).toBeDefined();
    expect(headers['strict-transport-security']).toContain('max-age=');
    // Referrer policy
    expect(headers['referrer-policy']).toBeDefined();
    // Permissions policy
    expect(headers['permissions-policy']).toBeDefined();
  });
  test('should not expose server information', async ({ request }) => {
    const response = await request.get('/');
    const headers = response.headers();
    expect(headers['server']).not.toContain('Apache');
    expect(headers['server']).not.toContain('nginx');
    expect(headers['x-powered-by']).toBeUndefined();
    expect(headers['x-aspnet-version']).toBeUndefined();
  });
  test('CORS should be restrictive', async ({ request }) => {
    const response = await request.get('/api/users', {
      headers: { Origin: 'https://evil.com' },
    });
    const corsHeader = response.headers()['access-control-allow-origin'];
    expect(corsHeader).not.toBe('*');
    if (corsHeader) {
      expect(corsHeader).not.toContain('evil.com');
    }
  });
  test('directory listing is disabled', async ({ request }) => {
    const response = await request.get('/static/');
    const body = await response.text();
    expect(body).not.toContain('Index of');
    expect(body).not.toContain('Directory listing');
  });
  test('debug endpoints are not exposed', async ({ request }) => {
    const debugPaths = [
      '/debug',
      '/actuator',
      '/actuator/env',
      '/_debug',
      '/api/debug',
      '/phpinfo.php',
      '/.env',
      '/wp-admin',
      '/graphql',  // Should require auth
    ];
    for (const path of debugPaths) {
      const response = await request.get(path);
      expect([401, 403, 404]).toContain(response.status());
    }
  });
});

import subprocess
import json
import time
from zapv2 import ZAPv2
class ZapSecurityScanner:
    def __init__(self, target_url: str, zap_api_key: str = ''):
        self.target = target_url
        self.zap = ZAPv2(apikey=zap_api_key, proxies={
            'http': 'http://127.0.0.1:8080',
            'https': 'http://127.0.0.1:8080',
        })
    def spider_scan(self, max_duration: int = 60):
        """Crawl the application to discover endpoints."""
        scan_id = self.zap.spider.scan(self.target)
        timeout = time.time() + max_duration
        while int(self.zap.spider.status(scan_id)) < 100:
            if time.time() > timeout:
                self.zap.spider.stop(scan_id)
                break
            time.sleep(2)
        return self.zap.spider.results(scan_id)
    def active_scan(self, max_duration: int = 300):
        """Run active security scan."""
        scan_id = self.zap.ascan.scan(self.target)
        timeout = time.time() + max_duration
        while int(self.zap.ascan.status(scan_id)) < 100:
            if time.time() > timeout:
                self.zap.ascan.stop(scan_id)
                break
            time.sleep(5)
        return scan_id