0
# Addon Development
1

2
Extensible addon system for custom functionality with lifecycle hooks, event handling, and access to all proxy components. Includes comprehensive built-in addons and framework for developing custom extensions.
3

4
## Capabilities
5

6
### Addon System Overview
7

8
The addon system provides a comprehensive framework for extending mitmproxy functionality.
9

10
```python { .api }
11
def default_addons() -> List[Any]:
12
    """
13
    Get list of all built-in addons.
14
    
15
    Returns:
16
    - List of addon instances providing core mitmproxy functionality
17
    """
18

19
def concurrent(func: Callable) -> Callable:
20
    """
21
    Decorator for concurrent addon execution.
22
    
23
    Allows addon methods to run concurrently without blocking
24
    the main proxy processing thread.
25
    
26
    Parameters:
27
    - func: Addon method to make concurrent
28
    
29
    Returns:
30
    - Decorated function that runs concurrently
31
    """
32
```
33

34
### Addon Context Access
35

36
Global context object providing access to proxy components.
37

38
```python { .api }
39
# Context module for addon access
40
import mitmproxy.ctx as ctx
41

42
# Context provides access to:
43
# - ctx.master: Master instance
44
# - ctx.options: Configuration options
45
# - ctx.log: Logging interface
46

47
class Context:
48
    """
49
    Global context object for addon access to proxy components.
50
    
51
    Available as `ctx` in addon methods.
52
    """
53
    master: Master
54
    options: Options
55
    log: LogEntry
56
    
57
    def log_info(self, message: str) -> None:
58
        """Log info message."""
59
    
60
    def log_warn(self, message: str) -> None:
61
        """Log warning message."""
62
    
63
    def log_error(self, message: str) -> None:
64
        """Log error message."""
65
```
66

67
### Built-in Addons
68

69
Comprehensive set of built-in addons for common functionality.
70

71
```python { .api }
72
# Core functionality addons
73
class Core:
74
    """Core proxy functionality."""
75

76
class Browser:
77
    """Browser integration and automation."""
78

79
class Block:
80
    """Request/response blocking capabilities."""
81

82
class BlockList:
83
    """Host and URL blacklisting functionality."""
84

85
class AntiCache:
86
    """Cache prevention for testing."""
87

88
class AntiComp:
89
    """Compression disabling for analysis."""
90

91
# Playback and testing addons
92
class ClientPlayback:
93
    """Client-side request replay."""
94

95
class ServerPlayback:
96
    """Server-side response replay."""
97

98
# Content modification addons
99
class ModifyBody:
100
    """Request/response body modification."""
101

102
class ModifyHeaders:
103
    """Header modification and injection."""
104

105
class MapRemote:
106
    """Remote URL mapping and redirection."""
107

108
class MapLocal:
109
    """Local file serving and mapping."""
110

111
# Authentication and security addons
112
class ProxyAuth:
113
    """Proxy authentication management."""
114

115
class StickyAuth:
116
    """Authentication persistence across requests."""
117

118
class StickyCookie:
119
    """Cookie persistence and management."""
120

121
class UpstreamAuth:
122
    """Upstream server authentication."""
123

124
# Data export and persistence addons
125
class Save:
126
    """Flow saving and persistence."""
127

128
class SaveHar:
129
    """HAR format export."""
130

131
class Export:
132
    """Multi-format flow export."""
133

134
# Network and protocol addons
135
class DnsResolver:
136
    """DNS resolution and caching."""
137

138
class NextLayer:
139
    """Protocol layer detection and routing."""
140

141
class TlsConfig:
142
    """TLS/SSL configuration management."""
143

144
# User interface addons
145
class CommandHistory:
146
    """Command history management."""
147

148
class Comment:
149
    """Flow commenting and annotation."""
150

151
class Cut:
152
    """Data extraction and filtering."""
153

154
# Development and debugging addons
155
class ScriptLoader:
156
    """Custom script loading and management."""
157

158
class Onboarding:
159
    """User onboarding and help system."""
160

161
class DisableH2C:
162
    """Disable HTTP/2 cleartext upgrade for compatibility."""
163

164
class StripDnsHttpsRecords:
165
    """Strip DNS HTTPS records for testing."""
166

167
class UpdateAltSvc:
168
    """Update Alt-Svc headers for HTTP/3 support."""
169
```
170

171
## Usage Examples
172

173
### Basic Addon Development
174

175
```python
176
from mitmproxy import http
177
import mitmproxy.ctx as ctx
178

179
class BasicAddon:
180
    """Basic addon example demonstrating common patterns."""
181
    
182
    def __init__(self):
183
        self.request_count = 0
184
        self.blocked_domains = {"malicious.com", "blocked.example"}
185
    
186
    def load(self, loader):
187
        """Called when the addon is loaded."""
188
        ctx.log.info("BasicAddon loaded")
189
        
190
        # Add custom options
191
        loader.add_option(
192
            name="basic_addon_enabled",
193
            typespec=bool,
194
            default=True,
195
            help="Enable basic addon functionality"
196
        )
197
    
198
    def configure(self, updates):
199
        """Called when configuration changes."""
200
        if "basic_addon_enabled" in updates:
201
            enabled = ctx.options.basic_addon_enabled
202
            ctx.log.info(f"BasicAddon {'enabled' if enabled else 'disabled'}")
203
    
204
    def request(self, flow: http.HTTPFlow) -> None:
205
        """Handle HTTP requests."""
206
        if not ctx.options.basic_addon_enabled:
207
            return
208
        
209
        self.request_count += 1
210
        
211
        # Log request
212
        ctx.log.info(f"Request #{self.request_count}: {flow.request.method} {flow.request.url}")
213
        
214
        # Block malicious domains
215
        if flow.request.host in self.blocked_domains:
216
            ctx.log.warn(f"Blocked request to {flow.request.host}")
217
            flow.response = http.Response.make(
218
                status_code=403,
219
                content="Blocked by BasicAddon",
220
                headers={"Content-Type": "text/plain"}
221
            )
222
            return
223
        
224
        # Add custom header
225
        flow.request.headers["X-BasicAddon"] = "processed"
226
    
227
    def response(self, flow: http.HTTPFlow) -> None:
228
        """Handle HTTP responses."""
229
        if not ctx.options.basic_addon_enabled:
230
            return
231
        
232
        if flow.response:
233
            # Add response header
234
            flow.response.headers["X-BasicAddon-Response"] = "processed"
235
            
236
            # Log response
237
            ctx.log.info(f"Response: {flow.response.status_code} for {flow.request.url}")
238
    
239
    def done(self):
240
        """Called when mitmproxy shuts down."""
241
        ctx.log.info(f"BasicAddon processed {self.request_count} requests")
242

243
# Register the addon
244
addons = [BasicAddon()]
245
```
246

247
### Advanced Addon with Concurrent Processing
248

249
```python
250
from mitmproxy import http
251
from mitmproxy.script import concurrent
252
import mitmproxy.ctx as ctx
253
import asyncio
254
import aiohttp
255
import json
256

257
class APIAnalyzerAddon:
258
    """Advanced addon with concurrent processing for API analysis."""
259
    
260
    def __init__(self):
261
        self.api_calls = []
262
        self.session = None
263
    
264
    def load(self, loader):
265
        """Initialize addon with configuration."""
266
        ctx.log.info("APIAnalyzerAddon loaded")
267
        
268
        # Add configuration options
269
        loader.add_option(
270
            name="api_analyzer_webhook",
271
            typespec=str,
272
            default="",
273
            help="Webhook URL for API analysis results"
274
        )
275
        
276
        loader.add_option(
277
            name="api_analyzer_filter",
278
            typespec=str,
279
            default="/api/",
280
            help="URL pattern to filter API calls"
281
        )
282
    
283
    async def setup_session(self):
284
        """Set up HTTP session for webhook calls."""
285
        if not self.session:
286
            self.session = aiohttp.ClientSession()
287
    
288
    @concurrent
289
    def request(self, flow: http.HTTPFlow) -> None:
290
        """Analyze API requests concurrently."""
291
        # Filter API calls
292
        filter_pattern = ctx.options.api_analyzer_filter
293
        if filter_pattern not in flow.request.url:
294
            return
295
        
296
        # Extract API call information
297
        api_call = {
298
            "timestamp": flow.request.timestamp_start,
299
            "method": flow.request.method,
300
            "url": flow.request.url,
301
            "headers": dict(flow.request.headers),
302
            "content_type": flow.request.headers.get("content-type", ""),
303
            "content_length": len(flow.request.content) if flow.request.content else 0
304
        }
305
        
306
        # Parse JSON request body
307
        if "application/json" in api_call["content_type"]:
308
            try:
309
                api_call["json_body"] = flow.request.json()
310
            except ValueError:
311
                api_call["json_body"] = None
312
        
313
        self.api_calls.append(api_call)
314
        ctx.log.info(f"Analyzed API call: {flow.request.method} {flow.request.url}")
315
    
316
    @concurrent
317
    def response(self, flow: http.HTTPFlow) -> None:
318
        """Analyze API responses concurrently."""
319
        filter_pattern = ctx.options.api_analyzer_filter
320
        if filter_pattern not in flow.request.url:
321
            return
322
        
323
        # Find corresponding request analysis
324
        for api_call in reversed(self.api_calls):
325
            if (api_call["url"] == flow.request.url and 
326
                api_call["method"] == flow.request.method):
327
                
328
                # Add response information
329
                if flow.response:
330
                    api_call.update({
331
                        "status_code": flow.response.status_code,
332
                        "response_headers": dict(flow.response.headers),
333
                        "response_content_type": flow.response.headers.get("content-type", ""),
334
                        "response_length": len(flow.response.content) if flow.response.content else 0,
335
                        "response_time": flow.response.timestamp_end - flow.request.timestamp_start if flow.response.timestamp_end else None
336
                    })
337
                    
338
                    # Parse JSON response
339
                    if "application/json" in api_call.get("response_content_type", ""):
340
                        try:
341
                            api_call["json_response"] = flow.response.json()
342
                        except ValueError:
343
                            api_call["json_response"] = None
344
                
345
                # Send to webhook if configured
346
                webhook_url = ctx.options.api_analyzer_webhook
347
                if webhook_url:
348
                    asyncio.create_task(self.send_webhook(api_call, webhook_url))
349
                
350
                break
351
    
352
    async def send_webhook(self, api_call_data, webhook_url):
353
        """Send API analysis data to webhook."""
354
        try:
355
            await self.setup_session()
356
            
357
            async with self.session.post(
358
                webhook_url,
359
                json=api_call_data,
360
                headers={"Content-Type": "application/json"}
361
            ) as response:
362
                if response.status == 200:
363
                    ctx.log.info(f"Webhook sent successfully for {api_call_data['url']}")
364
                else:
365
                    ctx.log.warn(f"Webhook failed with status {response.status}")
366
                    
367
        except Exception as e:
368
            ctx.log.error(f"Webhook error: {e}")
369
    
370
    def done(self):
371
        """Cleanup when addon shuts down."""
372
        if self.session:
373
            asyncio.create_task(self.session.close())
374
        
375
        ctx.log.info(f"APIAnalyzerAddon analyzed {len(self.api_calls)} API calls")
376

377
addons = [APIAnalyzerAddon()]
378
```
379

380
### Addon with State Management
381

382
```python
383
from mitmproxy import http
384
import mitmproxy.ctx as ctx
385
import sqlite3
386
import threading
387
from datetime import datetime
388

389
class TrafficLoggerAddon:
390
    """Addon that logs traffic to SQLite database with thread safety."""
391
    
392
    def __init__(self):
393
        self.db_path = "traffic_log.db"
394
        self.lock = threading.Lock()
395
        self.setup_database()
396
    
397
    def setup_database(self):
398
        """Initialize SQLite database."""
399
        with sqlite3.connect(self.db_path) as conn:
400
            conn.execute("""
401
                CREATE TABLE IF NOT EXISTS requests (
402
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
403
                    timestamp REAL,
404
                    method TEXT,
405
                    url TEXT,
406
                    host TEXT,
407
                    path TEXT,
408
                    status_code INTEGER,
409
                    request_size INTEGER,
410
                    response_size INTEGER,
411
                    response_time REAL,
412
                    user_agent TEXT,
413
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
414
                )
415
            """)
416
            conn.commit()
417
    
418
    def load(self, loader):
419
        """Configure addon options."""
420
        loader.add_option(
421
            name="traffic_logger_enabled",
422
            typespec=bool,
423
            default=True,
424
            help="Enable traffic logging to database"
425
        )
426
        
427
        loader.add_option(
428
            name="traffic_logger_filter",
429
            typespec=str,
430
            default="",
431
            help="Host filter for logging (empty = log all)"
432
        )
433
    
434
    def request(self, flow: http.HTTPFlow) -> None:
435
        """Log request initiation."""
436
        if not ctx.options.traffic_logger_enabled:
437
            return
438
        
439
        # Apply host filter
440
        host_filter = ctx.options.traffic_logger_filter
441
        if host_filter and host_filter not in flow.request.host:
442
            return
443
        
444
        # Store request start time for later response time calculation
445
        flow.metadata["log_start_time"] = datetime.now()
446
    
447
    def response(self, flow: http.HTTPFlow) -> None:
448
        """Log completed request/response."""
449
        if not ctx.options.traffic_logger_enabled:
450
            return
451
        
452
        # Apply host filter
453
        host_filter = ctx.options.traffic_logger_filter
454
        if host_filter and host_filter not in flow.request.host:
455
            return
456
        
457
        # Calculate response time
458
        response_time = None
459
        if "log_start_time" in flow.metadata:
460
            start_time = flow.metadata["log_start_time"]
461
            response_time = (datetime.now() - start_time).total_seconds()
462
        
463
        # Prepare log data
464
        log_data = {
465
            "timestamp": flow.request.timestamp_start,
466
            "method": flow.request.method,
467
            "url": flow.request.url,
468
            "host": flow.request.host,
469
            "path": flow.request.path,
470
            "status_code": flow.response.status_code if flow.response else None,
471
            "request_size": len(flow.request.content) if flow.request.content else 0,
472
            "response_size": len(flow.response.content) if flow.response and flow.response.content else 0,
473
            "response_time": response_time,
474
            "user_agent": flow.request.headers.get("User-Agent", "")
475
        }
476
        
477
        # Thread-safe database insert
478
        with self.lock:
479
            try:
480
                with sqlite3.connect(self.db_path) as conn:
481
                    conn.execute("""
482
                        INSERT INTO requests 
483
                        (timestamp, method, url, host, path, status_code, 
484
                         request_size, response_size, response_time, user_agent)
485
                        VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
486
                    """, (
487
                        log_data["timestamp"],
488
                        log_data["method"],
489
                        log_data["url"],
490
                        log_data["host"],
491
                        log_data["path"],
492
                        log_data["status_code"],
493
                        log_data["request_size"],
494
                        log_data["response_size"],
495
                        log_data["response_time"],
496
                        log_data["user_agent"]
497
                    ))
498
                    conn.commit()
499
                    
500
            except sqlite3.Error as e:
501
                ctx.log.error(f"Database error: {e}")
502
    
503
    def get_stats(self):
504
        """Get traffic statistics from database."""
505
        with self.lock:
506
            try:
507
                with sqlite3.connect(self.db_path) as conn:
508
                    cursor = conn.cursor()
509
                    
510
                    # Get basic stats
511
                    cursor.execute("SELECT COUNT(*) FROM requests")
512
                    total_requests = cursor.fetchone()[0]
513
                    
514
                    cursor.execute("SELECT COUNT(*) FROM requests WHERE status_code >= 400")
515
                    error_requests = cursor.fetchone()[0]
516
                    
517
                    cursor.execute("SELECT AVG(response_time) FROM requests WHERE response_time IS NOT NULL")
518
                    avg_response_time = cursor.fetchone()[0]
519
                    
520
                    return {
521
                        "total_requests": total_requests,
522
                        "error_requests": error_requests,
523
                        "avg_response_time": avg_response_time
524
                    }
525
                    
526
            except sqlite3.Error as e:
527
                ctx.log.error(f"Database error: {e}")
528
                return None
529
    
530
    def done(self):
531
        """Log final statistics."""
532
        stats = self.get_stats()
533
        if stats:
534
            ctx.log.info(f"TrafficLogger final stats: {stats}")
535

536
addons = [TrafficLoggerAddon()]
537
```
538

539
### Command-based Addon
540

541
```python
542
from mitmproxy import http, command
543
import mitmproxy.ctx as ctx
544
from typing import Sequence
545

546
class CommandAddon:
547
    """Addon that provides custom commands."""
548
    
549
    def __init__(self):
550
        self.blocked_hosts = set()
551
        self.request_count = 0
552
    
553
    @command.command("addon.block_host")
554
    def block_host(self, host: str) -> None:
555
        """Block requests to a specific host."""
556
        self.blocked_hosts.add(host)
557
        ctx.log.info(f"Blocked host: {host}")
558
    
559
    @command.command("addon.unblock_host")
560
    def unblock_host(self, host: str) -> None:
561
        """Unblock requests to a specific host."""
562
        self.blocked_hosts.discard(host)
563
        ctx.log.info(f"Unblocked host: {host}")
564
    
565
    @command.command("addon.list_blocked")
566
    def list_blocked(self) -> Sequence[str]:
567
        """List all blocked hosts."""
568
        return list(self.blocked_hosts)
569
    
570
    @command.command("addon.clear_blocked")
571
    def clear_blocked(self) -> None:
572
        """Clear all blocked hosts."""
573
        count = len(self.blocked_hosts)
574
        self.blocked_hosts.clear()
575
        ctx.log.info(f"Cleared {count} blocked hosts")
576
    
577
    @command.command("addon.stats")
578
    def get_stats(self) -> str:
579
        """Get addon statistics."""
580
        return f"Requests processed: {self.request_count}, Blocked hosts: {len(self.blocked_hosts)}"
581
    
582
    def request(self, flow: http.HTTPFlow) -> None:
583
        """Handle requests with blocking logic."""
584
        self.request_count += 1
585
        
586
        if flow.request.host in self.blocked_hosts:
587
            ctx.log.info(f"Blocked request to {flow.request.host}")
588
            flow.response = http.Response.make(
589
                status_code=403,
590
                content=f"Host {flow.request.host} is blocked",
591
                headers={"Content-Type": "text/plain"}
592
            )
593

594
addons = [CommandAddon()]
595
```
596

597
### Addon Integration Example
598

599
```python
600
# Example of loading and using custom addons
601

602
# Save addon code to files, e.g., basic_addon.py, api_analyzer.py, etc.
603

604
# Load addons in mitmproxy:
605
# 1. Command line: mitmproxy -s basic_addon.py -s api_analyzer.py
606
# 2. Programmatically:
607

608
from mitmproxy.tools.main import mitmdump
609
from mitmproxy import master, options
610
from basic_addon import BasicAddon
611
from api_analyzer import APIAnalyzerAddon
612

613
def run_with_addons():
614
    """Run mitmproxy with custom addons."""
615
    opts = options.Options(listen_port=8080)
616
    
617
    # Create master with addons
618
    m = master.Master(opts)
619
    m.addons.add(BasicAddon())
620
    m.addons.add(APIAnalyzerAddon())
621
    
622
    try:
623
        m.run()
624
    except KeyboardInterrupt:
625
        m.shutdown()
626

627
if __name__ == "__main__":
628
    run_with_addons()
629
```