HTTP server for phishing landing pages and tracking.
| 13 | from modules.phishing.templates import TemplateManager |
| 14 | |
| 15 | # Campaign ids are uuid4 strings and tracking ids are the first 8 characters of |
| 16 | # one, so anything outside this alphabet is not an id we issued. Rejecting the |
| 17 | # rest keeps path segments out of the HTML we render and out of the redirect |
| 18 | # Location header. |
| 19 | _ID_PATTERN = re.compile(r"\A[A-Za-z0-9_-]{1,64}\Z") |
| 20 | |
| 21 | |
| 22 | def _is_valid_id(value: str) -> bool: |
| 23 | """True if `value` looks like an id this server issued.""" |
| 24 | return bool(_ID_PATTERN.match(value)) |
| 25 | |
| 26 | |
| 27 | class PhishingServer: |
| 28 | """HTTP server for phishing landing pages and tracking.""" |
| 29 | |
| 30 | def __init__( |
| 31 | self, |
| 32 | host: str = "0.0.0.0", |
| 33 | port: int = 8080, |
| 34 | campaign_manager: CampaignManager | None = None, |
| 35 | template_manager: TemplateManager | None = None, |
| 36 | ): |
| 37 | self.host = host |
| 38 | self.port = port |
| 39 | self.logger = get_logger("phishing.server") |
| 40 | self.campaign_manager = campaign_manager or CampaignManager() |
| 41 | self.template_manager = template_manager or TemplateManager() |
| 42 | self.app: web.Application | None = None |
| 43 | self.runner: web.AppRunner | None = None |
| 44 | self._on_click_callbacks: list[Callable] = [] |
| 45 | self._on_submit_callbacks: list[Callable] = [] |
| 46 | |
| 47 | def on_click(self, callback: Callable) -> None: |
| 48 | """Register callback for link click events.""" |
| 49 | self._on_click_callbacks.append(callback) |
| 50 | |
| 51 | def on_submit(self, callback: Callable) -> None: |
| 52 | """Register callback for form submission events.""" |
| 53 | self._on_submit_callbacks.append(callback) |
| 54 | |
| 55 | async def start(self) -> None: |
| 56 | """Start the phishing server.""" |
| 57 | self.app = web.Application() |
| 58 | |
| 59 | # Routes |
| 60 | self.app.router.add_get("/track/{campaign_id}/{tracking_id}", self._handle_track_open) |
| 61 | self.app.router.add_get("/click/{campaign_id}/{tracking_id}", self._handle_click) |
| 62 | self.app.router.add_get("/landing/{campaign_id}/{tracking_id}", self._handle_landing) |
| 63 | self.app.router.add_post("/capture/{campaign_id}", self._handle_capture) |
| 64 | self.app.router.add_get("/pixel/{campaign_id}/{tracking_id}.gif", self._handle_pixel) |
| 65 | |
| 66 | self.runner = web.AppRunner(self.app) |
| 67 | await self.runner.setup() |
| 68 | |
| 69 | site = web.TCPSite(self.runner, self.host, self.port) |
| 70 | await site.start() |
| 71 | |
| 72 | self.logger.info(f"Phishing server started on http://{self.host}:{self.port}") |