Merge pull request #2 from enowars/feature/admin-in-checker

Feature/admin in checker

Author: inkustr

checker/Dockerfile

@@ -1,6 +1,22 @@
 FROM python:3.9-buster
 RUN apt-get update && apt-get upgrade -y 
 
+# Install Playwright system dependencies for Chromium
+RUN apt-get install -y \
+    libnss3 \
+    libnspr4 \
+    libdbus-1-3 \
+    libatk1.0-0 \
+    libatspi2.0-0 \
+    libxcomposite1 \
+    libxdamage1 \
+    libxfixes3 \
+    libxrandr2 \
+    libgbm1 \
+    libxkbcommon0 \
+    libasound2 \
+    && rm -rf /var/lib/apt/lists/*
+
 # add checker user
 RUN useradd -ms /bin/bash -u 1000 checker
 
@@ -10,6 +26,7 @@ COPY requirements.txt .
 
 USER checker
 RUN python3 -m pip install -r requirements.txt
+RUN python3 -m playwright install chromium
 
 COPY ./src .
 

checker/requirements.txt

@@ -2,4 +2,5 @@ enochecker3==0.8.1
 uvicorn
 gunicorn
 faker
-httpx
+httpx
+playwright

checker/src/admin_simulator.py

@@ -0,0 +1,117 @@
+import asyncio
+from typing import Optional
+from httpx import AsyncClient
+from logging import LoggerAdapter
+from enochecker3 import MumbleException
+from message_generator import generate_admin_message
+
+try:
+    from playwright.async_api import async_playwright
+    PLAYWRIGHT_AVAILABLE = True
+except ImportError:
+    PLAYWRIGHT_AVAILABLE = False
+
+
+class AdminSimulator:
+    def __init__(self, client: AsyncClient, logger: LoggerAdapter):
+        self.client = client
+        self.logger = logger
+        self.admin_password = "]!V$JuOzx@fi%pvG,lF!"
+        self.admin_username = "admin"
+        self.service_url = str(client.base_url)
+
+    def validate(self) -> None:
+        if not PLAYWRIGHT_AVAILABLE:
+            self.logger.warning("Playwright not available, skipping admin simulation")
+            return
+        
+        self.logger.info("Starting headless browser admin simulation...")
+
+    async def load_feedback_page(self) -> None:
+        self.validate()
+        
+        async with async_playwright() as p:
+            browser = await p.chromium.launch(
+                headless=True,
+                args=['--no-sandbox', '--disable-setuid-sandbox']
+            )
+            
+            try:
+                page = await browser.new_page()
+                
+                page.set_default_timeout(10000)
+                
+                await page.goto(f"{self.service_url}/")
+                
+                await page.fill('#login-username', self.admin_username)
+                await page.fill('#login-password', self.admin_password)
+                
+                await page.click('#login-form button[type="submit"]')
+                
+                await page.wait_for_load_state('networkidle')
+                
+                await page.goto(f"{self.service_url}/admin/feedback")
+                
+                await page.wait_for_load_state('networkidle')
+                
+                await asyncio.sleep(2)
+                
+                self.logger.info("Admin simulation completed - any XSS scripts would have executed")
+                
+            except Exception as e:
+                self.logger.warning(f"Admin simulation failed: {e}")
+                raise MumbleException(f"Admin simulation error: {e}")
+            finally:
+                await browser.close()
+
+    async def post_new_message(self) -> None:
+        self.validate()
+
+        message_data = generate_admin_message()
+        message_text = message_data["message"]
+        message_year = message_data["year"]
+        
+        self.logger.info(f"Posting time traveller message from {message_year}: {message_text[:50]}...")
+
+        async with async_playwright() as p:
+            browser = await p.chromium.launch(
+                headless=True,
+                args=['--no-sandbox', '--disable-setuid-sandbox']
+            )
+            
+            try:
+                page = await browser.new_page()
+                
+                page.set_default_timeout(10000)
+                
+                await page.goto(f"{self.service_url}/")
+                
+                await page.fill('#login-username', self.admin_username)
+                await page.fill('#login-password', self.admin_password)
+                
+                await page.click('#login-form button[type="submit"]')
+                
+                await page.wait_for_load_state('networkidle')
+                
+                await page.goto(f"{self.service_url}/admin/message")
+                
+                await page.wait_for_load_state('networkidle')
+                
+                await page.fill('#year', str(message_year))
+                await page.fill('#message', message_text)
+                
+                await page.click('button[type="submit"]')
+                
+                await page.wait_for_load_state('networkidle')
+                
+                await asyncio.sleep(2)
+                
+                self.logger.info(f"Admin message posted successfully from year {message_year}")
+                
+            except Exception as e:
+                self.logger.warning(f"Admin message posting failed: {e}")
+                raise MumbleException(f"Admin message posting error: {e}")
+            finally:
+                await browser.close()
+
+

checker/src/checker.py

@@ -28,6 +28,7 @@
     AsyncSocket,
 )
 from enochecker3.utils import assert_equals, assert_in
+import admin_simulator
 
 """
 Checker config
@@ -518,5 +519,19 @@ async def havoc_feedback_image(task: HavocCheckerTaskMessage, client: AsyncClien
         logger.warning("Simple SVG feedback does not appear on feedback page")
         raise MumbleException("Can't submit feedback with SVG image")
 
+@checker.havoc(1)
+async def havoc_admin_simulation(task: HavocCheckerTaskMessage, client: AsyncClient, logger: LoggerAdapter) -> None:
+    logger.info("Starting admin simulation havoc test...")
+
+    simulator = admin_simulator.AdminSimulator(client, logger)
+    
+    try:
+        await simulator.load_feedback_page()
+        await simulator.post_new_message()
+        logger.info("Admin simulation completed successfully")
+        
+    except Exception as e:
+        raise MumbleException(f"Admin could not post message or visit feedback page.")
+
 if __name__ == "__main__":
     checker.run()