Spaces:

paras9909
/

opticparse-python

Paused

File size: 13,835 Bytes

bcf46c3

import urllib.request
import ssl
import time
import json
import uuid
import os

print('\n=== SECTION 3: Generate Test API Key ===')
SUPABASE_URL = 'https://xxmvhvxeglsjbewlouqg.supabase.co'
SERVICE_KEY = 'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpc3MiOiJzdXBhYmFzZSIsInJlZiI6Inh4bXZodnhlZ2xzamJld2xvdXFnIiwicm9sZSI6InNlcnZpY2Vfcm9sZSIsImlhdCI6MTc4MjkwMjUxMSwiZXhwIjoyMDk4NDc4NTExfQ.3-uJVjh8FDP8dLar2A6pLRs1K0O5I8lyZepZk7NizJU'

email = f'verify_{uuid.uuid4().hex[:6]}@test.com'
req = urllib.request.Request(
    f'{SUPABASE_URL}/auth/v1/admin/users',
    method='POST',
    headers={
        'Content-Type': 'application/json',
        'apikey': SERVICE_KEY,
        'Authorization': f'Bearer {SERVICE_KEY}'
    },
    data=json.dumps({'email': email, 'password': 'VerifyTest123!', 'email_confirm': True}).encode()
)
PYTHON = 'https://opticparse-python-sg.onrender.com'
NODE = 'https://opticparse-1opticparse-node-sg.onrender.com'
try:
    res = urllib.request.urlopen(req)
    user_id = json.loads(res.read())['id']

    req2 = urllib.request.Request(
        f'{PYTHON}/gateway/keys/generate',
        method='POST',
        headers={'Content-Type': 'application/json'},
        data=json.dumps({'user_id': user_id}).encode()
    )
    res2 = urllib.request.urlopen(req2, timeout=30)
    api_key = json.loads(res2.read())['api_key']
    print(f'✅ API key generated: {api_key[:20]}...')
    with open('verify_key.txt', 'w') as f:
        f.write(f'{user_id}|{api_key}')
except Exception as e:
    print('❌ API key generation failed:', e)
    user_id = ''
    api_key = ''

print('\n=== SECTION 4: Test All API Endpoints ===')
def test_endpoint(name, method, url, data=None, headers=None, expected_codes=[200]):
    try:
        start = time.time()
        h = {'Content-Type': 'application/json', 'X-API-Key': api_key}
        if headers: h.update(headers)
        if headers and 'X-API-Key' in headers and headers['X-API-Key'] is None:
            del h['X-API-Key']
        req = urllib.request.Request(url, method=method, data=data, headers=h)
        res = urllib.request.urlopen(req, timeout=90)
        elapsed = time.time() - start
        body = res.read().decode()[:200]
        status = '✅' if res.status in expected_codes else '⚠️'
        print(f'{status} {name}: {res.status} ({elapsed:.1f}s)')
        return res.status, body
    except urllib.error.HTTPError as e:
        elapsed = time.time() - start
        body = e.read().decode()[:200]
        status = '✅' if e.code in expected_codes else '❌'
        print(f'{status} {name}: {e.code} ({elapsed:.1f}s)')
        return e.code, body
    except Exception as e:
        print(f'❌ {name}: {str(e)[:80]}')
        return 0, str(e)

if user_id:
    test_endpoint('Gateway - Generate Key', 'POST', f'{PYTHON}/gateway/keys/generate', json.dumps({'user_id': user_id}).encode())
    test_endpoint('Gateway - Get Usage', 'GET', f'{PYTHON}/gateway/usage/{user_id}')
    test_endpoint('Gateway - Regenerate Key', 'POST', f'{PYTHON}/gateway/keys/regenerate', json.dumps({'user_id': user_id}).encode())
    test_endpoint('NEW - Usage History (30 days)', 'GET', f'{PYTHON}/gateway/usage/{user_id}/history')

    test_endpoint('Vision Scrape (sync)', 'POST', f'{PYTHON}/api/vision-scrape', json.dumps({'target_url': 'https://news.ycombinator.com', 'extraction_query': 'Top 3 post titles', 'response_schema': {'posts': [{'title': 'string'}]}}).encode())

    status, body = test_endpoint('NEW - Vision Scrape (async)', 'POST', f'{PYTHON}/api/vision-scrape/async', json.dumps({'target_url': 'https://example.com', 'extraction_query': 'Main heading', 'response_schema': {'heading': 'string'}}).encode())
    job_id = None
    try: job_id = json.loads(body).get('job_id')
    except: pass

    time.sleep(3)
    if job_id: test_endpoint('NEW - Job Status Check', 'GET', f'{PYTHON}/api/vision-scrape/jobs/{job_id}')

    status, body = test_endpoint('Watch - Create', 'POST', f'{PYTHON}/api/watch', json.dumps({'target_url': 'https://example.com', 'extraction_query': 'Monitor changes'}).encode())
    watch_id = None
    try: watch_id = json.loads(body).get('watch_id') or json.loads(body).get('id')
    except: pass
    if watch_id: test_endpoint('Watch - Get Diff', 'GET', f'{PYTHON}/api/watch/{watch_id}/diff')

test_endpoint('PhishVision - Single Detect', 'POST', f'{NODE}/api/phish-detect', json.dumps({'url': 'https://google.com'}).encode(), headers={'X-API-Key': None})
test_endpoint('PhishVision - Bulk Scan', 'POST', f'{NODE}/api/phish-batch-simple', json.dumps({'urls': ['https://google.com', 'https://github.com']}).encode(), headers={'X-API-Key': None})
test_endpoint('PhishVision - PDF Report', 'GET', f'{NODE}/api/phish-report?url=https://example.com', headers={'X-API-Key': None}, expected_codes=[200])
test_endpoint('PhishVision - Create Monitor', 'POST', f'{NODE}/api/monitor', json.dumps({'url': 'https://example.com', 'webhook_url': 'https://example.com/webhook'}).encode(), headers={'X-API-Key': None})
test_endpoint('PhishVision - Cache Stats', 'GET', f'{NODE}/api/cache-stats', headers={'X-API-Key': None})

print('\n=== SECTION 8: New Features Verification ===')
if user_id:
    try:
        req = urllib.request.Request(f'{PYTHON}/gateway/usage/{user_id}/history', headers={'X-API-Key': api_key})
        res = urllib.request.urlopen(req, timeout=30)
        data = json.loads(res.read())
        days = len(data.get('history', []))
        print(f'✅ Usage history: returns {days} days of data')
    except Exception as e: print(f'❌ Usage history: {str(e)[:80]}')

    try:
        req = urllib.request.Request(f'{PYTHON}/api/vision-scrape/async', method='POST', headers={'Content-Type': 'application/json', 'X-API-Key': api_key}, data=json.dumps({'target_url': 'https://example.com', 'extraction_query': 'What is the heading?', 'response_schema': {'heading': 'string'}}).encode())
        res = urllib.request.urlopen(req, timeout=30)
        data = json.loads(res.read())
        job_id = data.get('job_id')
        print(f'✅ Async scrape: job_id={job_id[:8]}... status={data.get("status")}')
        print('   Waiting 20s for job to complete...')
        time.sleep(20)
        req2 = urllib.request.Request(f'{PYTHON}/api/vision-scrape/jobs/{job_id}', headers={'X-API-Key': api_key})
        res2 = urllib.request.urlopen(req2, timeout=30)
        job_data = json.loads(res2.read())
        job_status = job_data.get('status')
        if job_status == 'completed': print(f'   ✅ Async job completed!')
        elif job_status == 'processing': print(f'   ⚠️  Still processing (normal for first run)')
        else: print(f'   ❌ Job failed: {job_data.get("error")}')
    except Exception as e: print(f'❌ Async scrape: {str(e)[:80]}')

try:
    req = urllib.request.Request(f'{NODE}/api/phish-detect', method='POST', headers={'Content-Type': 'application/json'}, data=json.dumps({'url': 'https://github.com'}).encode())
    res = urllib.request.urlopen(req, timeout=90)
    data = json.loads(res.read())
    required_fields = ['verdict', 'confidence_score_percentage', 'domain_age_days', 'registrar', 'redirect_chain', 'visual_anomalies_detected']
    missing = [f for f in required_fields if f not in data]
    if not missing: print(f'✅ PhishVision full JSON: all fields present')
    else: print(f'❌ PhishVision missing fields: {missing}')
except Exception as e: print(f'❌ PhishVision full JSON: {str(e)[:80]}')

print('\n=== SECTION 5: Security Tests ===')
try:
    req = urllib.request.Request(f'{PYTHON}/api/vision-scrape', method='POST', headers={'Content-Type': 'application/json', 'X-API-Key': 'fake-key-123'}, data=json.dumps({'target_url': 'https://example.com', 'extraction_query': 'test'}).encode())
    urllib.request.urlopen(req, timeout=10)
    print('❌ Fake key NOT rejected')
except urllib.error.HTTPError as e:
    if e.code == 401: print('✅ Fake key rejected (401)')
    else: print(f'⚠️  Fake key got {e.code}')
except Exception as e:
    print(f'⚠️  {str(e)[:60]}')

try:
    req = urllib.request.Request(f'{NODE}/api/phish-detect', method='POST', headers={'Content-Type': 'application/json'}, data=json.dumps({'url': 'http://127.0.0.1/secret'}).encode())
    urllib.request.urlopen(req, timeout=10)
    print('❌ Internal URL NOT blocked')
except urllib.error.HTTPError as e:
    if e.code in [400, 403]: print(f'✅ Internal URL blocked ({e.code})')
    else: print(f'⚠️  Internal URL got {e.code}')
except Exception as e:
    print(f'⚠️  {str(e)[:60]}')

try:
    req = urllib.request.Request(f'{PYTHON}/api/vision-scrape', method='POST', headers={'Content-Type': 'application/json', 'X-API-Key': 'wrong-format-key'}, data=json.dumps({'target_url': 'https://example.com', 'extraction_query': 'test'}).encode())
    urllib.request.urlopen(req, timeout=10)
    print('❌ Wrong format key NOT rejected')
except urllib.error.HTTPError as e:
    if e.code == 401: print('✅ Wrong format key rejected (401)')
    else: print(f'⚠️  Wrong format key got {e.code}')
except Exception as e:
    print(f'⚠️  {str(e)[:60]}')

try:
    req = urllib.request.Request('https://opticparse.com', headers={'User-Agent': 'Mozilla/5.0'})
    res = urllib.request.urlopen(req, timeout=15)
    print(f'✅ HTTPS working: {res.status}')
except Exception as e:
    print(f'❌ HTTPS: {str(e)[:60]}')

print('\n=== SECTION 6: Database Security ===')
anon_key = ''
try:
    with open('dashboard/.env.production') as f:
        for line in f:
            if 'VITE_SUPABASE_ANON_KEY' in line:
                anon_key = line.split('=', 1)[1].strip().strip('"').strip("'")
except: pass
tables = ['users', 'api_keys', 'usage_logs', 'watches', 'monitors']
for table in tables:
    try:
        req = urllib.request.Request(f'{SUPABASE_URL}/rest/v1/{table}?select=*', headers={'apikey': anon_key, 'Authorization': f'Bearer {anon_key}'})
        res = urllib.request.urlopen(req, timeout=10)
        data = json.loads(res.read())
        if len(data) == 0: print(f'✅ {table}: RLS enforced')
        else: print(f'❌ {table}: BREACH — {len(data)} rows exposed')
    except: print(f'✅ {table}: RLS enforced')

print('\n=== SECTION 7: Content Verification ===')
files = {
    'docs/index.html': [('No GPT-4o', 'GPT-4o', False), ('WebLLM section', 'WebLLM', True), ('About page link', 'about.html', True), ('Privacy linked', 'privacy', True), ('Terms linked', 'terms', True), ('Dashboard link', 'dashboard.opticparse.com', True), ('LemonSqueezy URL', 'lemonsqueezy.com', True), ('Schema accuracy', 'Schema Accuracy', True), ('No placeholder', 'PLACEHOLDER', False), ('No old URL', 'parastejpal987', False), ('Async feature', 'async', True), ('Bulk scanning', 'Bulk', True)],
    'docs/about.html': [('About page exists', 'OpticParse', True), ('Punjab mentioned', 'Punjab', True), ('WebLLM mentioned', 'WebLLM', True), ('Contact email', 'support@opticparse.com', True)],
    'docs/privacy.html': [('Contact email', 'support@opticparse.com', True), ('No placeholder email', 'your email', False)],
    'docs/terms.html': [('Contact email', 'support@opticparse.com', True), ('No placeholder email', 'your email', False)]
}
for filepath, checks in files.items():
    if not os.path.exists(filepath): print(f'❌ {filepath}: FILE NOT FOUND'); continue
    content = open(filepath, 'r', encoding='utf-8').read().lower()
    for name, term, should_exist in checks:
        found = term.lower() in content
        print(f'{"✅" if (found == should_exist) else "❌"} {name}')

print('\n=== SECTION 8: New Features Verification ===')
if user_id:
    try:
        req = urllib.request.Request(f'{PYTHON}/gateway/usage/{user_id}/history', headers={'X-API-Key': api_key})
        res = urllib.request.urlopen(req, timeout=30)
        data = json.loads(res.read())
        days = len(data.get('history', []))
        print(f'✅ Usage history: returns {days} days of data')
    except Exception as e: print(f'❌ Usage history: {str(e)[:80]}')

    try:
        req = urllib.request.Request(f'{PYTHON}/api/vision-scrape/async', method='POST', headers={'Content-Type': 'application/json', 'X-API-Key': api_key}, data=json.dumps({'target_url': 'https://example.com', 'extraction_query': 'What is the heading?', 'response_schema': {'heading': 'string'}}).encode())
        res = urllib.request.urlopen(req, timeout=30)
        data = json.loads(res.read())
        job_id = data.get('job_id')
        print(f'✅ Async scrape: job_id={job_id[:8]}... status={data.get("status")}')
        print('   Waiting 20s for job to complete...')
        time.sleep(20)
        req2 = urllib.request.Request(f'{PYTHON}/api/vision-scrape/jobs/{job_id}', headers={'X-API-Key': api_key})
        res2 = urllib.request.urlopen(req2, timeout=30)
        job_data = json.loads(res2.read())
        job_status = job_data.get('status')
        if job_status == 'completed': print(f'   ✅ Async job completed!')
        elif job_status == 'processing': print(f'   ⚠️  Still processing (normal for first run)')
        else: print(f'   ❌ Job failed: {job_data.get("error")}')
    except Exception as e: print(f'❌ Async scrape: {str(e)[:80]}')

try:
    req = urllib.request.Request(f'{NODE}/api/phish-detect', method='POST', headers={'Content-Type': 'application/json'}, data=json.dumps({'url': 'https://github.com'}).encode())
    res = urllib.request.urlopen(req, timeout=90)
    data = json.loads(res.read())
    required_fields = ['verdict', 'confidence_score_percentage', 'domain_age_days', 'registrar', 'redirect_chain', 'visual_anomalies_detected']
    missing = [f for f in required_fields if f not in data]
    if not missing: print(f'✅ PhishVision full JSON: all fields present')
    else: print(f'❌ PhishVision missing fields: {missing}')
except Exception as e: print(f'❌ PhishVision full JSON: {str(e)[:80]}')