-
Notifications
You must be signed in to change notification settings - Fork 0
API Reference
Dragon edited this page Dec 22, 2025
·
1 revision
Last updated: 12/22/2025
This API reference documents the public interfaces for extending NaviDuck. Use these APIs to build plugins, custom search engines, UI themes, and integrations.
┌─────────────────────────────────────────────────────────────┐
│ Extension Points │
├─────────────────────────────────────────────────────────────┤
│ 🔍 Search Engines │ 🤖 AI Plugins │ 🎨 UI Themes │
│ 🔧 Core Hooks │ 📡 Network Layer │ 💾 Data Store │
│ 🛡️ Security │ ⚡ Performance │ 🔌 Integrations│
└─────────────────────────────────────────────────────────────┘
The central state management system that tracks all browser data and configuration.
class BrowserState:
# Configuration
use_emoji: bool # Icon theme preference
current_engine: str # Active search engine ID
tor_enabled: bool # Tor status
# Data Stores
history: List[Dict] # Browsing history
bookmarks: List[Dict] # Saved bookmarks
current_results: List[Dict] # Latest search results
# Session State
current_page: str # Current page content
current_url: str # Current URL
current_title: str # Current page title
# File Paths
data_file: str # User data file path
config_file: str # Configuration file path# Data Management
def load_data(self) -> None:
"""Load user data from disk."""
def save_data(self) -> None:
"""Save user data to disk."""
def load_config(self) -> None:
"""Load configuration from disk."""
def save_config(self) -> None:
"""Save configuration to disk."""
# Bookmark Management
def add_bookmark(self, title: str, url: str) -> bool:
"""Add a bookmark. Returns success status."""
def remove_bookmark(self, url: str) -> bool:
"""Remove a bookmark by URL."""
def get_bookmark(self, url: str) -> Optional[Dict]:
"""Retrieve bookmark by URL."""
# History Management
def add_history(self, entry: Dict) -> None:
"""Add entry to history."""
def clear_history(self, older_than_days: int = None) -> int:
"""Clear history. Returns number of entries removed."""
# Configuration
def update_config(self, key: str, value: Any) -> None:
"""Update configuration value."""
def get_config(self, key: str, default: Any = None) -> Any:
"""Retrieve configuration value."""# Subscribe to state changes
state.add_listener(event_type, callback)
# Event Types
EVENT_SEARCH_COMPLETE = "search_complete"
EVENT_PAGE_LOADED = "page_loaded"
EVENT_BOOKMARK_ADDED = "bookmark_added"
EVENT_HISTORY_UPDATED = "history_updated"
EVENT_CONFIG_CHANGED = "config_changed"
EVENT_TOR_STATUS_CHANGED = "tor_status_changed"
# Example: Listen for search completion
def on_search_complete(results):
print(f"Search returned {len(results)} results")
state.add_listener(EVENT_SEARCH_COMPLETE, on_search_complete)Base class for implementing custom search engines.
class SearchEngine(ABC):
"""Abstract base class for search engines."""
@abstractmethod
def search(self, query: str) -> List[SearchResult]:
"""Execute search and return results."""
pass
@abstractmethod
def parse_response(self, response: Response) -> List[SearchResult]:
"""Parse HTTP response into structured results."""
pass
@property
@abstractmethod
def name(self) -> str:
"""Display name of search engine."""
pass
@property
@abstractmethod
def icon(self) -> str:
"""Icon identifier for display."""
pass
@property
def supports_tor(self) -> bool:
"""Whether engine supports Tor routing."""
return False
@property
def rate_limit(self) -> Tuple[int, int]:
"""Rate limit: (requests, seconds)."""
return (10, 60)class SearchResult:
def __init__(
self,
title: str,
url: str,
snippet: str = "",
engine: str = "custom",
metadata: Dict = None
):
self.title = title
self.url = url
self.snippet = snippet
self.engine = engine
self.metadata = metadata or {}
def to_dict(self) -> Dict:
"""Convert to dictionary format."""
return {
"title": self.title[:100], # Truncated for display
"url": self.url,
"snippet": self.snippet[:150], # Truncated
"engine": self.engine,
"metadata": self.metadata
}# Custom search engine implementation
class MySearchEngine(SearchEngine):
def __init__(self):
self.base_url = "https://api.example.com/search"
self.api_key = None
def search(self, query: str) -> List[SearchResult]:
params = {"q": query, "format": "json"}
if self.api_key:
params["key"] = self.api_key
response = requests.get(self.base_url, params=params)
return self.parse_response(response)
def parse_response(self, response: Response) -> List[SearchResult]:
data = response.json()
results = []
for item in data.get("results", []):
result = SearchResult(
title=item.get("title", ""),
url=item.get("url", ""),
snippet=item.get("description", ""),
engine="mysearch",
metadata={
"score": item.get("score", 0),
"date": item.get("date", "")
}
)
results.append(result)
return results
@property
def name(self) -> str:
return "My Search"
@property
def icon(self) -> str:
return "MY_ICON"
# Register with SearchManager
search_manager.register_engine("mysearch", MySearchEngine())# engines/custom_engines.json
{
"mysearch": {
"name": "My Search Engine",
"url": "https://api.example.com/search",
"params": {
"q": "{query}",
"format": "json",
"api_key": "${API_KEY}"
},
"parser": "json", # json, html, xml, custom
"requires_tor": false,
"icon": "SEARCH",
"headers": {
"User-Agent": "NaviDuck/1.0"
},
"rate_limit": {
"requests": 10,
"seconds": 60
}
}
}
# Load configuration
search_manager.load_engines_from_config("engines/custom_engines.json")@search_engine(
name="My Search",
icon="MY_ICON",
supports_tor=True
)
def my_search_engine(query: str, context: Dict) -> List[Dict]:
"""Custom search engine function."""
# Implementation
return results
# Engine is automatically registeredFor custom HTML/XML parsing logic.
class SearchParser:
"""Base class for custom parsers."""
def parse_html(self, html: str, query: str = "") -> List[SearchResult]:
"""Parse HTML content."""
# Default implementation using regex
pattern = r'<a[^>]+href="([^"]+)"[^>]*>([^<]+)</a>'
matches = re.findall(pattern, html)
results = []
for url, title in matches[:10]:
if self._is_valid_result(url, title):
results.append(SearchResult(
title=title.strip(),
url=url,
snippet=self._extract_snippet(html, title),
engine="custom"
))
return results
def parse_json(self, data: Dict, query: str = "") -> List[SearchResult]:
"""Parse JSON response."""
# Implementation depends on API structure
pass
def _is_valid_result(self, url: str, title: str) -> bool:
"""Filter out invalid/spam results."""
return (len(title) > 10 and
not url.startswith("#") and
not any(spam in url for spam in self._spam_domains))
def _extract_snippet(self, html: str, title: str) -> str:
"""Extract relevant snippet around title."""
# Simple implementation
idx = html.find(title)
if idx != -1:
start = max(0, idx - 50)
end = min(len(html), idx + len(title) + 100)
return html[start:end]
return ""Intercept and modify search behavior.
class SearchHook:
"""Hook into search process."""
def pre_search(self, query: str, engine: str, context: Dict) -> Optional[Tuple[str, str]]:
"""
Called before search execution.
Return (modified_query, modified_engine) to alter search,
or None to proceed normally.
"""
# Example: Auto-complete queries
if len(query) < 3:
return None # Don't modify
# Add site: filter for programming queries
if any(word in query.lower() for word in ["python", "javascript", "code"]):
return f"{query} site:stackoverflow.com", engine
return None
def post_search(self, query: str, engine: str, results: List[SearchResult]) -> List[SearchResult]:
"""
Called after search results are retrieved.
Modify or filter results.
"""
# Example: Filter out low-quality results
filtered = [
r for r in results
if self._is_high_quality(r)
]
# Example: Add metadata
for result in filtered:
result.metadata["processed"] = True
return filtered
def on_error(self, query: str, engine: str, error: Exception) -> Optional[List[SearchResult]]:
"""
Called when search fails.
Return fallback results or None to propagate error.
"""
# Example: Provide cached results
cached = self._get_cached_results(query)
if cached:
return cached
# Example: Try alternative query
simplified = self._simplify_query(query)
return search_manager.search(simplified, "ddg")
def _is_high_quality(self, result: SearchResult) -> bool:
"""Quality heuristic."""
return (
len(result.title) > 15 and
len(result.snippet) > 20 and
"http" in result.url and
not any(spam in result.url for spam in ["ad", "track", "click"])
)# Create hook instance
quality_hook = QualityFilterHook()
# Register with search manager
search_manager.add_hook(quality_hook)
# Multiple hooks execute in registration order
search_manager.add_hook(SpamFilterHook())
search_manager.add_hook(ResultRankerHook())Extend NavAI with custom knowledge sources and response generators.
class AIPlugin(ABC):
"""Base class for AI plugins."""
def __init__(self, context: Dict = None):
self.context = context or {}
self.priority = 50 # 0-100, higher = tried first
@abstractmethod
def can_handle(self, query: str) -> bool:
"""
Determine if this plugin can handle the query.
Return True to attempt processing.
"""
pass
@abstractmethod
def process(self, query: str) -> Optional[str]:
"""
Process query and return response.
Return None if cannot answer.
"""
pass
def get_confidence(self, query: str) -> float:
"""
Confidence score (0-1) for handling this query.
Used when multiple plugins can handle same query.
"""
return 0.5
def learn_from_feedback(self, query: str, response: str, was_helpful: bool) -> None:
"""Learn from user feedback."""
passclass ProgrammingHelpPlugin(AIPlugin):
"""Provides programming-related answers."""
def __init__(self):
super().__init__()
self.priority = 80 # High priority for programming questions
self.knowledge_base = self._load_knowledge()
def can_handle(self, query: str) -> bool:
query_lower = query.lower()
programming_keywords = [
"python", "javascript", "java", "code", "programming",
"function", "class", "variable", "loop", "array",
"how to", "syntax", "error", "bug", "debug"
]
return any(keyword in query_lower for keyword in programming_keywords)
def process(self, query: str) -> Optional[str]:
# Check local knowledge base
answer = self._check_knowledge_base(query)
if answer:
return answer
# Fall back to code snippet generation
if "example" in query.lower() or "code" in query.lower():
return self._generate_example(query)
return None # Let other plugins handle
def _load_knowledge(self) -> Dict[str, str]:
"""Load programming knowledge."""
return {
"what is a variable": "A variable is a named storage location...",
"python list comprehension": "List comprehensions provide a concise way...",
# ... more knowledge
}
def _check_knowledge_base(self, query: str) -> Optional[str]:
"""Check if query matches known patterns."""
for pattern, answer in self.knowledge_base.items():
if pattern in query.lower():
return answer
return None
def _generate_example(self, query: str) -> str:
"""Generate code example based on query."""
if "python" in query.lower():
return "```python\n# Example: Hello World\nprint('Hello, World!')\n```"
elif "javascript" in query.lower():
return "```javascript\n// Example: Hello World\nconsole.log('Hello, World!');\n```"
return "Here's an example related to your query..."class WeatherPlugin(AIPlugin):
"""Provides weather information."""
def __init__(self, api_key: str = None):
super().__init__()
self.api_key = api_key
self.priority = 70
def can_handle(self, query: str) -> bool:
weather_terms = ["weather", "temperature", "forecast", "rain", "sunny"]
return any(term in query.lower() for term in weather_terms)
def process(self, query: str) -> Optional[str]:
# Extract location
location = self._extract_location(query)
if not location:
return "Please specify a location (e.g., 'weather in London')"
# Get weather data
weather_data = self._fetch_weather(location)
if not weather_data:
return f"Could not fetch weather for {location}"
# Format response
return self._format_weather_response(weather_data)
def _extract_location(self, query: str) -> Optional[str]:
"""Extract location from query."""
patterns = [
r"weather in (.+)",
r"temperature in (.+)",
r"forecast for (.+)"
]
for pattern in patterns:
match = re.search(pattern, query.lower())
if match:
return match.group(1).strip()
return None
def _fetch_weather(self, location: str) -> Optional[Dict]:
"""Fetch weather data from API."""
if not self.api_key:
return None
try:
response = requests.get(
"https://api.weatherapi.com/v1/current.json",
params={
"key": self.api_key,
"q": location,
"aqi": "no"
}
)
return response.json()
except:
return None
def _format_weather_response(self, data: Dict) -> str:
"""Format weather data into readable response."""
location = data["location"]["name"]
temp_c = data["current"]["temp_c"]
condition = data["current"]["condition"]["text"]
return f"Weather in {location}: {temp_c}°C, {condition}"# Create plugin instances
programming_plugin = ProgrammingHelpPlugin()
weather_plugin = WeatherPlugin(api_key="your_api_key")
# Register with NavAI
navai = NavAI()
navai.register_plugin(programming_plugin)
navai.register_plugin(weather_plugin)
# Or load from configuration
navai.load_plugins_from_config("plugins/ai_plugins.json")Customize how AI responses are displayed.
class AIResponseFormatter:
"""Format AI responses for display."""
def format(self, response: str, query: str = "") -> str:
"""Format raw response for display."""
# Default formatting
formatted = response.strip()
# Add icons/colors based on content
if "error" in response.lower():
formatted = f"❌ {formatted}"
elif "success" in response.lower():
formatted = f"✅ {formatted}"
# Wrap long lines
formatted = self._wrap_text(formatted, width=70)
return formatted
def format_with_context(self, response: str, context: Dict) -> str:
"""Format response with additional context."""
# Add source attribution
if context.get("source"):
response = f"{response}\n\nSource: {context['source']}"
# Add confidence score
if confidence := context.get("confidence"):
response = f"{response}\n\nConfidence: {confidence:.0%}"
return self.format(response)
def _wrap_text(self, text: str, width: int) -> str:
"""Wrap text to specified width."""
import textwrap
return "\n".join(textwrap.wrap(text, width=width))Create custom color schemes and icon sets.
class Theme:
"""UI theme with colors and icons."""
def __init__(self, name: str):
self.name = name
self.colors = self._default_colors()
self.icons = self._default_icons()
def _default_colors(self) -> Dict[str, str]:
"""Default color scheme."""
return {
# Text colors
"text.primary": "\033[37m", # White
"text.secondary": "\033[90m", # Gray
"text.success": "\033[32m", # Green
"text.error": "\033[31m", # Red
"text.warning": "\033[33m", # Yellow
"text.info": "\033[36m", # Cyan
"text.highlight": "\033[35m", # Magenta
# Background colors
"bg.primary": "\033[40m", # Black background
"bg.secondary": "\033[47m", # White background
"bg.highlight": "\033[45m", # Magenta background
# Special
"reset": "\033[0m", # Reset all
"bold": "\033[1m", # Bold
"underline": "\033[4m", # Underline
}
def _default_icons(self) -> Dict[str, str]:
"""Default icon set."""
return {
"search": "🔍",
"ai": "🤖",
"bookmark": "📑",
"history": "🕰️",
"tor": "🧅",
"settings": "⚙️",
"error": "❌",
"success": "✅",
"warning": "⚠️",
"info": "ℹ️",
}
def apply(self) -> None:
"""Apply theme to UI."""
UIManager.current_theme = self
def get_color(self, key: str, default: str = "") -> str:
"""Get color code by key."""
return self.colors.get(key, default)
def get_icon(self, key: str, default: str = "?") -> str:
"""Get icon by key."""
return self.icons.get(key, default)class DarkTheme(Theme):
"""Dark mode theme."""
def __init__(self):
super().__init__("dark")
self.colors.update({
"text.primary": "\033[37m", # Bright white
"text.secondary": "\033[90m", # Dark gray
"bg.primary": "\033[40m", # Black
"bg.secondary": "\033[48;5;236m",# Dark gray
})
self.icons.update({
"search": "", # Nerd Font icons
"ai": "",
"bookmark": "",
})class LightTheme(Theme):
"""Light mode theme."""
def __init__(self):
super().__init__("light")
self.colors.update({
"text.primary": "\033[30m", # Black
"text.secondary": "\033[90m", # Dark gray
"bg.primary": "\033[47m", # White
"bg.secondary": "\033[48;5;255m",# Light gray
})class HighContrastTheme(Theme):
"""High contrast theme for accessibility."""
def __init__(self):
super().__init__("high-contrast")
self.colors.update({
"text.primary": "\033[97m", # Bright white
"text.secondary": "\033[37m", # White
"text.success": "\033[92m", # Bright green
"text.error": "\033[91m", # Bright red
"bg.primary": "\033[40m", # Black
"bg.secondary": "\033[100m", # Bright black
})
self.icons.update({
"search": "🔍",
"ai": "🤖",
# Larger, clearer icons
})class IconSet:
"""Collection of icons for different contexts."""
def __init__(self, name: str):
self.name = name
self.icons = {}
def add_icon(self, key: str, icon: str, variants: Dict[str, str] = None):
"""Add icon with optional variants."""
self.icons[key] = {
"default": icon,
"variants": variants or {}
}
def get_icon(self, key: str, variant: str = "default") -> str:
"""Get icon by key and variant."""
if key not in self.icons:
return "?"
icon_data = self.icons[key]
return icon_data["variants"].get(variant, icon_data["default"])
def get_icon_for_context(self, key: str, context: Dict) -> str:
"""Get icon based on context (theme, platform, etc.)."""
icon = self.get_icon(key)
# Adjust based on context
if context.get("theme") == "dark":
# Use brighter icons for dark themes
pass
if context.get("platform") == "windows":
# Use Windows-friendly icons
pass
return iconclass NerdFontIconSet(IconSet):
"""Nerd Font icon set."""
def __init__(self):
super().__init__("nerd-font")
self.add_icon("search", "")
self.add_icon("ai", "")
self.add_icon("bookmark", "")
self.add_icon("history", "")
self.add_icon("tor", "")
self.add_icon("settings", "")
self.add_icon("error", "")
self.add_icon("success", "")
self.add_icon("warning", "")
self.add_icon("info", "")class EmojiIconSet(IconSet):
"""Emoji icon set."""
def __init__(self):
super().__init__("emoji")
self.add_icon("search", "🔍")
self.add_icon("ai", "🤖")
self.add_icon("bookmark", "📑")
self.add_icon("history", "🕰️")
self.add_icon("tor", "🧅")
self.add_icon("settings", "⚙️")
self.add_icon("error", "❌")
self.add_icon("success", "✅")
self.add_icon("warning", "⚠️")
self.add_icon("info", "ℹ️")# Create themes
dark_theme = DarkTheme()
light_theme = LightTheme()
high_contrast = HighContrastTheme()
# Register with theme manager
theme_manager = ThemeManager()
theme_manager.register_theme(dark_theme)
theme_manager.register_theme(light_theme)
theme_manager.register_theme(high_contrast)
# Apply theme
theme_manager.apply_theme("dark")
# Or load from configuration
theme_manager.load_themes_from_dir("themes/")Hook into various points in NaviDuck's execution.
class Hook:
"""Base class for all hooks."""
def __init__(self, name: str, priority: int = 50):
self.name = name
self.priority = priority # Execution order: 0-100
def get_hook_points(self) -> List[str]:
"""
Return list of hook points this hook subscribes to.
Example: ["pre_command", "post_search", "on_error"]
"""
return []# Command Processing
HOOK_PRE_COMMAND = "pre_command" # Before command execution
HOOK_POST_COMMAND = "post_command" # After command execution
HOOK_COMMAND_ERROR = "command_error" # On command error
# Search Process
HOOK_PRE_SEARCH = "pre_search" # Before search
HOOK_POST_SEARCH = "post_search" # After search
HOOK_SEARCH_ERROR = "search_error" # On search error
# Page Loading
HOOK_PRE_PAGE_LOAD = "pre_page_load" # Before page load
HOOK_POST_PAGE_LOAD = "post_page_load" # After page load
HOOK_PAGE_ERROR = "page_error" # On page load error
# AI Processing
HOOK_PRE_AI_QUERY = "pre_ai_query" # Before AI query
HOOK_POST_AI_RESPONSE = "post_ai_response" # After AI response
HOOK_AI_ERROR = "ai_error" # On AI error
# Data Management
HOOK_PRE_DATA_SAVE = "pre_data_save" # Before saving data
HOOK_POST_DATA_LOAD = "post_data_load" # After loading data
# UI Events
HOOK_PRE_UI_RENDER = "pre_ui_render" # Before UI render
HOOK_POST_UI_RENDER = "post_ui_render" # After UI render
# System Events
HOOK_STARTUP = "startup" # On startup
HOOK_SHUTDOWN = "shutdown" # On shutdown
HOOK_ERROR = "error" # Any errorclass AnalyticsHook(Hook):
"""Track usage analytics."""
def __init__(self):
super().__init__("analytics", priority=10)
self.events = []
def get_hook_points(self) -> List[str]:
return [
HOOK_PRE_COMMAND,
HOOK_POST_SEARCH,
HOOK_POST_AI_RESPONSE,
HOOK_ERROR
]
def on_pre_command(self, command: str, args: List[str]) -> None:
"""Track command execution."""
self.events.append({
"timestamp": time.time(),
"type": "command",
"command": command,
"args": args
})
def on_post_search(self, query: str, engine: str, results_count: int) -> None:
"""Track search performance."""
self.events.append({
"timestamp": time.time(),
"type": "search",
"query": query,
"engine": engine,
"results": results_count
})
def on_error(self, error: Exception, context: Dict) -> None:
"""Track errors."""
self.events.append({
"timestamp": time.time(),
"type": "error",
"error": str(error),
"context": context
})
def save_analytics(self) -> None:
"""Save analytics data."""
import json
with open("analytics.json", "w") as f:
json.dump(self.events, f, indent=2)class SecurityHook(Hook):
"""Security monitoring and validation."""
def __init__(self):
super().__init__("security", priority=5) # High priority
def get_hook_points(self) -> List[str]:
return [
HOOK_PRE_COMMAND,
HOOK_PRE_PAGE_LOAD,
HOOK_PRE_DATA_SAVE
]
def on_pre_command(self, command: str, args: List[str]) -> Optional[Tuple[str, List[str]]]:
"""Validate and sanitize commands."""
# Block dangerous commands
dangerous = ["rm", "format", "delete"]
if any(cmd in command.lower() for cmd in dangerous):
raise SecurityError(f"Blocked dangerous command: {command}")
# Sanitize arguments
sanitized_args = [self._sanitize(arg) for arg in args]
return command, sanitized_args
def on_pre_page_load(self, url: str) -> Optional[str]:
"""Validate URLs before loading."""
# Check for malicious patterns
if self._is_malicious_url(url):
raise SecurityError(f"Blocked malicious URL: {url}")
# Enforce HTTPS for sensitive sites
if "login" in url or "bank" in url:
if not url.startswith("https://"):
url = url.replace("http://", "https://")
return url
def _sanitize(self, text: str) -> str:
"""Sanitize input text."""
# Remove control characters
return ''.join(char for char in text if ord(char) >= 32)
def _is_malicious_url(self, url: str) -> bool:
"""Check if URL is potentially malicious."""
malicious_patterns = [
r"javascript:",
r"data:text/html",
r"file://",
r"\.exe$",
r"\.bat$",
r"\.sh$"
]
for pattern in malicious_patterns:
if re.search(pattern, url, re.IGNORECASE):
return True
return False# Create hook instances
analytics_hook = AnalyticsHook()
security_hook = SecurityHook()
logging_hook = LoggingHook()
# Register with hook manager
hook_manager = HookManager()
hook_manager.register_hook(analytics_hook)
hook_manager.register_hook(security_hook)
hook_manager.register_hook(logging_hook)
# Hooks execute in priority order (lowest first)
# Security (5) → Analytics (10) → Logging (50)class CustomHTTPAdapter(requests.adapters.HTTPAdapter):
"""Custom HTTP adapter with additional features."""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.request_hooks = []
self.response_hooks = []
def send(self, request, **kwargs):
"""Override send with custom logic."""
# Pre-request hooks
for hook in self.request_hooks:
request = hook(request)
# Add custom headers
request.headers.update({
"X-NaviDuck-Version": "1.0",
"X-Request-ID": str(uuid.uuid4())
})
# Execute request
response = super().send(request, **kwargs)
# Post-response hooks
for hook in self.response_hooks:
response = hook(response)
return response
def add_request_hook(self, hook: Callable):
"""Add hook to modify requests."""
self.request_hooks.append(hook)
def add_response_hook(self, hook: Callable):
"""Add hook to modify responses."""
self.response_hooks.append(hook)class ProxyManager:
"""Manage proxy configurations."""
def __init__(self):
self.proxies = {}
self.current_proxy = None
def add_proxy(self, name: str, config: Dict):
"""Add proxy configuration."""
self.proxies[name] = {
"http": config.get("http"),
"https": config.get("https"),
"socks": config.get("socks"),
"auth": config.get("auth"), # (username, password)
"enabled": config.get("enabled", True)
}
def set_proxy(self, name: str) -> bool:
"""Set active proxy."""
if name not in self.proxies:
return False
proxy_config = self.proxies[name]
if not proxy_config["enabled"]:
return False
self.current_proxy = proxy_config
return True
def get_proxy_config(self) -> Dict:
"""Get current proxy configuration for requests."""
if not self.current_proxy:
return {}
config = {}
proxy = self.current_proxy
if proxy.get("http"):
config["http"] = self._format_proxy_url("http", proxy)
if proxy.get("https"):
config["https"] = self._format_proxy_url("https", proxy)
if proxy.get("socks"):
config["http"] = proxy["socks"]
config["https"] = proxy["socks"]
return config
def _format_proxy_url(self, scheme: str, proxy: Dict) -> str:
"""Format proxy URL with authentication."""
url = proxy[scheme]
if auth := proxy.get("auth"):
username, password = auth
url = url.replace("://", f"://{username}:{password}@")
return urlclass SSLConfig:
"""Configure SSL/TLS settings."""
def __init__(self):
self.verify = True
self.cert = None
self.ciphers = None
self.min_version = None
self.max_version = None
self.ca_cert = None
def create_context(self) -> ssl.SSLContext:
"""Create SSL context with configuration."""
context = ssl.create_default_context()
if self.ciphers:
context.set_ciphers(self.ciphers)
if self.min_version:
context.minimum_version = self.min_version
if self.max_version:
context.maximum_version = self.max_version
if self.ca_cert:
context.load_verify_locations(self.ca_cert)
return context
def disable_verification(self):
"""Disable SSL verification (not recommended for production)."""
self.verify = False
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)# Create network components
proxy_manager = ProxyManager()
ssl_config = SSLConfig()
adapter = CustomHTTPAdapter()
# Configure
proxy_manager.add_proxy("tor", {
"socks": "socks5h://127.0.0.1:9050",
"enabled": True
})
ssl_config.ciphers = "ECDHE+AESGCM:ECDHE+CHACHA20"
ssl_config.min_version = ssl.TLSVersion.TLSv1_2
# Apply to session
session = requests.Session()
session.mount("http://", adapter)
session.mount("https://", adapter)
# Set proxy
session.proxies = proxy_manager.get_proxy_config()
session.verify = ssl_config.create_context()class StorageBackend(ABC):
"""Abstract storage backend."""
@abstractmethod
def save(self, key: str, data: Any) -> bool:
"""Save data with key."""
pass
@abstractmethod
def load(self, key: str) -> Any:
"""Load data by key."""
pass
@abstractmethod
def delete(self, key: str) -> bool:
"""Delete data by key."""
pass
@abstractmethod
def exists(self, key: str) -> bool:
"""Check if key exists."""
pass
@abstractmethod
def list_keys(self, pattern: str = "*") -> List[str]:
"""List keys matching pattern."""
passclass JSONStorageBackend(StorageBackend):
"""JSON file storage backend."""
def __init__(self, base_path: str = "~/.naviduck"):
self.base_path = os.path.expanduser(base_path)
os.makedirs(self.base_path, exist_ok=True)
def save(self, key: str, data: Any) -> bool:
"""Save data as JSON."""
try:
filepath = self._get_filepath(key)
with open(filepath, "w", encoding="utf-8") as f:
json.dump(data, f, indent=2, ensure_ascii=False)
return True
except:
return False
def load(self, key: str) -> Any:
"""Load data from JSON."""
try:
filepath = self._get_filepath(key)
if not os.path.exists(filepath):
return None
with open(filepath, "r", encoding="utf-8") as f:
return json.load(f)
except:
return None
def _get_filepath(self, key: str) -> str:
"""Convert key to filepath."""
# Sanitize key for filename
safe_key = "".join(c for c in key if c.isalnum() or c in "._-")
return os.path.join(self.base_path, f"{safe_key}.json")class SQLiteStorageBackend(StorageBackend):
"""SQLite storage backend."""
def __init__(self, db_path: str = "~/.naviduck/data.db"):
self.db_path = os.path.expanduser(db_path)
self._init_db()
def _init_db(self):
"""Initialize database schema."""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS storage (
key TEXT PRIMARY KEY,
value TEXT,
created TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
conn.commit()
conn.close()
def save(self, key: str, data: Any) -> bool:
"""Save data to SQLite."""
try:
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
value = json.dumps(data)
cursor.execute("""
INSERT OR REPLACE INTO storage (key, value, updated)
VALUES (?, ?, CURRENT_TIMESTAMP)
""", (key, value))
conn.commit()
conn.close()
return True
except:
return Falseclass EncryptedStorageBackend(StorageBackend):
"""Encrypted storage backend."""
def __init__(self, backend: StorageBackend, encryption_key: bytes):
self.backend = backend
self.cipher = Fernet(encryption_key)
def save(self, key: str, data: Any) -> bool:
"""Save encrypted data."""
try:
# Serialize and encrypt
serialized = json.dumps(data).encode()
encrypted = self.cipher.encrypt(serialized)
# Save through backend
return self.backend.save(key, encrypted.hex())
except:
return False
def load(self, key: str) -> Any:
"""Load and decrypt data."""
try:
# Load encrypted data
encrypted_hex = self.backend.load(key)
if not encrypted_hex:
return None
# Decrypt and deserialize
encrypted = bytes.fromhex(encrypted_hex)
decrypted = self.cipher.decrypt(encrypted)
return json.loads(decrypted.decode())
except:
return Noneclass CachedStorageBackend(StorageBackend):
"""Storage backend with caching."""
def __init__(self, backend: StorageBackend, max_cache_size: int = 1000):
self.backend = backend
self.cache = {}
self.cache_order = [] # For LRU
self.max_cache_size = max_cache_size
def save(self, key: str, data: Any) -> bool:
"""Save to cache and backend."""
# Update cache
self._update_cache(key, data)
# Save to backend
return self.backend.save(key, data)
def load(self, key: str) -> Any:
"""Load from cache if available."""
# Check cache first
if key in self.cache:
# Move to front (most recently used)
self.cache_order.remove(key)
self.cache_order.insert(0, key)
return self.cache[key]
# Load from backend
data = self.backend.load(key)
if data is not None:
self._update_cache(key, data)
return data
def _update_cache(self, key: str, data: Any):
"""Update cache with LRU eviction."""
if key in self.cache:
self.cache_order.remove(key)
elif len(self.cache) >= self.max_cache_size:
# Evict least recently used
lru_key = self.cache_order.pop()
del self.cache[lru_key]
self.cache[key] = data
self.cache_order.insert(0, key)# Create backend chain
json_backend = JSONStorageBackend()
encrypted_backend = EncryptedStorageBackend(json_backend, encryption_key)
cached_backend = CachedStorageBackend(encrypted_backend)
# Use with BrowserState
state = BrowserState(storage_backend=cached_backend)
# Or register multiple backends
storage_manager = StorageManager()
storage_manager.register_backend("json", json_backend)
storage_manager.register_backend("encrypted", encrypted_backend)
storage_manager.register_backend("cached", cached_backend)
# Use specific backend
storage_manager.use_backend("cached")
storage_manager.save("history", history_data)class InputValidator:
"""Validate and sanitize user input."""
def validate_search_query(self, query: str) -> Tuple[bool, str]:
"""Validate search query."""
if not query or not query.strip():
return False, "Query cannot be empty"
if len(query) > 500:
return False, "Query too long (max 500 characters)"
# Check for dangerous patterns
dangerous = ["<script>", "javascript:", "onload="]
for pattern in dangerous:
if pattern in query.lower():
return False, "Query contains unsafe content"
return True, query.strip()
def validate_url(self, url: str) -> Tuple[bool, str]:
"""Validate URL."""
try:
result = urlparse(url)
# Check scheme
if result.scheme not in ["http", "https", "ftp"]:
return False, "Unsupported URL scheme"
# Check for localhost/internal
if self._is_local_address(result.netloc):
return False, "Local addresses not allowed"
# Check for malicious patterns
if self._is_malicious_url(url):
return False, "URL appears malicious"
return True, url
except:
return False, "Invalid URL format"
def sanitize_html(self, html: str) -> str:
"""Sanitize HTML content."""
# Remove scripts
html = re.sub(r'<script[^>]*>.*?</script>', '', html, flags=re.DOTALL)
# Remove event handlers
html = re.sub(r'on\w+="[^"]*"', '', html)
html = re.sub(r"on\w+='[^']*'", '', html)
# Remove dangerous protocols
html = re.sub(r'javascript:', '', html, flags=re.IGNORECASE)
html = re.sub(r'data:', '', html, flags=re.IGNORECASE)
return htmlclass RateLimiter:
"""Rate limiting for API calls."""
def __init__(self, max_requests: int, period: int):
self.max_requests = max_requests
self.period = period
self.requests = defaultdict(list)
def check(self, key: str) -> Tuple[bool, float]:
"""
Check if request is allowed.
Returns (allowed, wait_time)
"""
now = time.time()
# Clean old requests
self.requests[key] = [
timestamp for timestamp in self.requests[key]
if now - timestamp < self.period
]
# Check limit
if len(self.requests[key]) >= self.max_requests:
oldest = self.requests[key][0]
wait_time = self.period - (now - oldest)
return False, wait_time
# Add request
self.requests[key].append(now)
return True, 0
def get_remaining(self, key: str) -> int:
"""Get remaining requests in period."""
now = time.time()
self.requests[key] = [
timestamp for timestamp in self.requests[key]
if now - timestamp < self.period
]
return self.max_requests - len(self.requests[key])class CertificatePinner:
"""SSL certificate pinning."""
def __init__(self):
self.pins = {} # hostname -> [sha256 hashes]
def add_pin(self, hostname: str, certificate_hash: str):
"""Add certificate pin for hostname."""
if hostname not in self.pins:
self.pins[hostname] = []
self.pins[hostname].append(certificate_hash)
def verify(self, hostname: str, certificate: bytes) -> bool:
"""Verify certificate matches pinned hash."""
if hostname not in self.pins:
return True # No pin, allow
# Calculate hash
cert_hash = hashlib.sha256(certificate).digest()
b64_hash = base64.b64encode(cert_hash).decode()
# Check against pins
return b64_hash in self.pins[hostname]
def create_adapter(self) -> requests.adapters.HTTPAdapter:
"""Create HTTP adapter with pinning."""
class PinningAdapter(requests.adapters.HTTPAdapter):
def __init__(self, pinner):
super().__init__()
self.pinner = pinner
def cert_verify(self, conn, url, verify, cert):
super().cert_verify(conn, url, verify, cert)
hostname = urlparse(url).hostname
peer_cert = conn.sock.getpeercert(binary_form=True)
if not self.pinner.verify(hostname, peer_cert):
raise requests.exceptions.SSLError(
f"Certificate pinning violation for {hostname}"
)
return PinningAdapter(self)# Setup security
validator = InputValidator()
rate_limiter = RateLimiter(max_requests=10, period=60)
cert_pinner = CertificatePinner()
# Add certificate pins
cert_pinner.add_pin("api.duckduckgo.com", "sha256/...")
cert_pinner.add_pin("wikipedia.org", "sha256/...")
# Create secure session
session = requests.Session()
session.mount("https://", cert_pinner.create_adapter())
# Validate input
is_valid, message = validator.validate_search_query(user_input)
if not is_valid:
print(f"Invalid input: {message}")class Profiler:
"""Performance profiling."""
def __init__(self):
self.measurements = {}
self.start_times = {}
def start(self, name: str):
"""Start measurement."""
self.start_times[name] = time.perf_counter()
def stop(self, name: str) -> float:
"""Stop measurement and return duration."""
if name not in self.start_times:
return 0.0
duration = time.perf_counter() - self.start_times[name]
if name not in self.measurements:
self.measurements[name] = []
self.measurements[name].append(duration)
return duration
def get_stats(self, name: str) -> Dict:
"""Get statistics for measurement."""
if name not in self.measurements:
return {}
times = self.measurements[name]
return {
"count": len(times),
"total": sum(times),
"avg": sum(times) / len(times),
"min": min(times),
"max": max(times),
"last": times[-1] if times else 0
}
def print_report(self):
"""Print performance report."""
print("Performance Report:")
print("=" * 50)
for name in sorted(self.measurements.keys()):
stats = self.get_stats(name)
print(f"{name}:")
print(f" Count: {stats['count']}")
print(f" Average: {stats['avg']:.3f}s")
print(f" Min/Max: {stats['min']:.3f}s / {stats['max']:.3f}s")
print()class CacheManager:
"""Unified cache management."""
def __init__(self):
self.caches = {}
def register_cache(self, name: str, cache: Any):
"""Register cache instance."""
self.caches[name] = cache
def get_cache(self, name: str) -> Optional[Any]:
"""Get cache by name."""
return self.caches.get(name)
def clear_all(self):
"""Clear all caches."""
for cache in self.caches.values():
if hasattr(cache, "clear"):
cache.clear()
def get_stats(self) -> Dict:
"""Get cache statistics."""
stats = {}
for name, cache in self.caches.items():
if hasattr(cache, "get_stats"):
stats[name] = cache.get_stats()
else:
stats[name] = {"size": len(cache) if hasattr(cache, "__len__") else "unknown"}
return statsclass MemoryMonitor:
"""Monitor memory usage."""
def __init__(self):
self.samples = []
self.max_samples = 100
def sample(self) -> float:
"""Take memory sample."""
import psutil
import os
process = psutil.Process(os.getpid())
memory_mb = process.memory_info().rss / 1024 / 1024
self.samples.append(memory_mb)
if len(self.samples) > self.max_samples:
self.samples.pop(0)
return memory_mb
def get_stats(self) -> Dict:
"""Get memory statistics."""
if not self.samples:
return {}
return {
"current": self.samples[-1],
"average": sum(self.samples) / len(self.samples),
"min": min(self.samples),
"max": max(self.samples),
"trend": self._calculate_trend()
}
def _calculate_trend(self) -> str:
"""Calculate memory trend."""
if len(self.samples) < 10:
return "unknown"
recent = self.samples[-5:]
older = self.samples[-10:-5]
avg_recent = sum(recent) / len(recent)
avg_older = sum(older) / len(older)
if avg_recent > avg_older * 1.1:
return "increasing"
elif avg_recent < avg_older * 0.9:
return "decreasing"
else:
return "stable"
def check_for_leak(self) -> bool:
"""Check for potential memory leak."""
if len(self.samples) < 20:
return False
# Check if memory consistently increasing
segments = [
self.samples[:5],
self.samples[5:10],
self.samples[10:15],
self.samples[15:]
]
averages = [sum(seg) / len(seg) for seg in segments]
# Check if each segment is larger than previous
for i in range(1, len(averages)):
if averages[i] < averages[i-1] * 1.05: # Less than 5% increase
return False
return True# Setup performance monitoring
profiler = Profiler()
cache_manager = CacheManager()
memory_monitor = MemoryMonitor()
# Profile operations
profiler.start("search")
results = search_manager.search("test")
profiler.stop("search")
# Monitor memory
memory_monitor.sample()
# Check for issues
if memory_monitor.check_for_leak():
print("⚠️ Potential memory leak detected!")
# Get performance report
profiler.print_report()
print(f"Memory: {memory_monitor.get_stats()}")class WebhookManager:
"""Manage webhook integrations."""
def __init__(self):
self.webhooks = {}
def register_webhook(self, event: str, url: str, secret: str = None):
"""Register webhook for event."""
if event not in self.webhooks:
self.webhooks[event] = []
self.webhooks[event].append({
"url": url,
"secret": secret,
"enabled": True
})
def trigger(self, event: str, data: Dict):
"""Trigger webhooks for event."""
if event not in self.webhooks:
return
for webhook in self.webhooks[event]:
if not webhook["enabled"]:
continue
self._send_webhook(webhook, event, data)
def _send_webhook(self, webhook: Dict, event: str, data: Dict):
"""Send webhook request."""
payload = {
"event": event,
"timestamp": time.time(),
"data": data
}
headers = {
"User-Agent": "NaviDuck/1.0",
"Content-Type": "application/json"
}
# Add signature if secret provided
if webhook["secret"]:
import hmac
import hashlib
payload_str = json.dumps(payload)
signature = hmac.new(
webhook["secret"].encode(),
payload_str.encode(),
hashlib.sha256
).hexdigest()
headers["X-NaviDuck-Signature"] = signature
try:
response = requests.post(
webhook["url"],
json=payload,
headers=headers,
timeout=5
)
response.raise_for_status()
except:
pass # Log error but don't crashclass CLIWrapper:
"""Wrap NaviDuck for CLI integration."""
def __init__(self, naviduck_instance):
self.naviduck = naviduck_instance
def execute_command(self, command: str) -> str:
"""Execute command and return output."""
import io
import sys
# Capture output
old_stdout = sys.stdout
sys.stdout = io.StringIO()
try:
# Execute command
self.naviduck.handle_command(command)
# Get output
output = sys.stdout.getvalue()
return output.strip()
finally:
sys.stdout = old_stdout
def search_and_return(self, query: str, engine: str = None) -> List[Dict]:
"""Search and return results as data."""
results = self.naviduck.search_manager.search(query, engine)
return [r.to_dict() for r in results]
def ask_ai(self, question: str) -> str:
"""Ask AI and return response."""
return self.naviduck.navai.ask(question)class RESTServer:
"""REST API server for NaviDuck."""
def __init__(self, naviduck_instance, host="127.0.0.1", port=8080):
self.naviduck = naviduck_instance
self.host = host
self.port = port
from flask import Flask, request, jsonify
self.app = Flask(__name__)
self._setup_routes()
def _setup_routes(self):
"""Setup REST API routes."""
@self.app.route("/api/search", methods=["POST"])
def search():
data = request.json
results = self.naviduck.search_manager.search(
data["query"],
data.get("engine")
)
return jsonify([r.to_dict() for r in results])
@self.app.route("/api/ai", methods=["POST"])
def ai():
data = request.json
response = self.naviduck.navai.ask(data["query"])
return jsonify({"response": response})
@self.app.route("/api/bookmarks", methods=["GET", "POST"])
def bookmarks():
if request.method == "GET":
return jsonify(self.naviduck.state.bookmarks)
else:
data = request.json
self.naviduck.state.add_bookmark(
data["title"],
data["url"]
)
return jsonify({"success": True})
def start(self):
"""Start REST server."""
self.app.run(host=self.host, port=self.port){
"name": "my-plugin",
"version": "1.0.0",
"description": "My custom plugin",
"author": "Your Name",
"license": "MIT",
"entry_points": {
"search_engines": "my_plugin.search_engine:MySearchEngine",
"ai_plugins": "my_plugin.ai_plugin:MyAIPlugin",
"themes": "my_plugin.theme:MyTheme",
"hooks": "my_plugin.hooks:MyHook"
},
"dependencies": [
"requests>=2.25.0",
"beautifulsoup4>=4.9.0"
],
"config_schema": {
"api_key": {
"type": "string",
"description": "API key for the service",
"required": false
},
"enabled": {
"type": "boolean",
"default": true,
"description": "Enable the plugin"
}
}
}class PluginLoader:
"""Load and manage plugins."""
def __init__(self):
self.plugins = {}
self.loaded_plugins = {}
def load_plugin(self, path: str) -> bool:
"""Load plugin from path."""
try:
# Read manifest
manifest_path = os.path.join(path, "plugin.json")
with open(manifest_path, "r") as f:
manifest = json.load(f)
# Load entry points
for entry_type, entry_point in manifest["entry_points"].items():
self._load_entry_point(manifest, entry_type, entry_point)
# Store plugin
plugin_id = f"{manifest['name']}@{manifest['version']}"
self.loaded_plugins[plugin_id] = {
"manifest": manifest,
"path": path,
"loaded_at": time.time()
}
return True
except Exception as e:
print(f"Failed to load plugin {path}: {e}")
return False
def _load_entry_point(self, manifest: Dict, entry_type: str, entry_point: str):
"""Load specific entry point."""
module_name, class_name = entry_point.split(":")
# Import module
import importlib
module = importlib.import_module(module_name)
# Get class
plugin_class = getattr(module, class_name)
# Instantiate with config
config = manifest.get("config", {})
plugin_instance = plugin_class(**config)
# Register based on type
if entry_type == "search_engines":
search_manager.register_engine(
manifest["name"],
plugin_instance
)
elif entry_type == "ai_plugins":
navai.register_plugin(plugin_instance)
elif entry_type == "themes":
theme_manager.register_theme(plugin_instance)
elif entry_type == "hooks":
hook_manager.register_hook(plugin_instance)
def list_plugins(self) -> List[Dict]:
"""List loaded plugins."""
return [
{
"id": plugin_id,
"name": info["manifest"]["name"],
"version": info["manifest"]["version"],
"loaded_at": info["loaded_at"]
}
for plugin_id, info in self.loaded_plugins.items()
]# my_search_engine.py
from naviduck.api import SearchEngine, SearchResult
class MySearchEngine(SearchEngine):
def search(self, query: str):
# Simple implementation
return [
SearchResult(
title="Example Result",
url="https://example.com",
snippet="This is an example search result",
engine="mysearch"
)
]
@property
def name(self):
return "My Search"
@property
def icon(self):
return "🔍"
# Register
search_manager.register_engine("mysearch", MySearchEngine())# my_ai_plugin.py
from naviduck.api import AIPlugin
class GreetingPlugin(AIPlugin):
def can_handle(self, query: str):
return query.lower() in ["hello", "hi", "hey"]
def process(self, query: str):
return "Hello! I'm your custom AI plugin."
# Register
navai.register_plugin(GreetingPlugin())# Project structure
my-naviduck-extension/
├── plugin.json # Manifest
├── __init__.py
├── search_engine.py # Search engine implementation
├── ai_plugin.py # AI plugin
└── README.md
# Install locally
pip install -e .
# Or package for distribution
python setup.py sdist bdist_wheel- Keep Dependencies Minimal: Only require what's necessary
- Handle Errors Gracefully: Don't crash NaviDuck
- Respect User Privacy: Don't collect data without consent
- Follow Style Guidelines: Match NaviDuck's code style
- Provide Documentation: Document your extension
- Test Thoroughly: Ensure compatibility
- Version Your Extensions: Use semantic versioning
- Cache expensive operations
- Use async for I/O operations when possible
- Clean up resources properly
- Monitor memory usage
- Validate all input
- Sanitize output
- Use environment variables for secrets
- Implement rate limiting
- Follow least privilege principle
- Check existing extensions for examples
- Review the source code
- Ask in GitHub Discussions
- Create an issue for bug reports
# Test script
from my_extension import MySearchEngine
def test_extension():
engine = MySearchEngine()
results = engine.search("test")
assert len(results) > 0
print("✅ Extension works!")
if __name__ == "__main__":
test_extension()# Enable debug logging
import logging
logging.basicConfig(level=logging.DEBUG)
# Use the profiler
profiler.start("my_extension")
# ... your code ...
profiler.stop("my_extension")
profiler.print_report()Last updated: 12/22/2025
API Reference version: 3.0
Ready to extend NaviDuck? Start with a simple plugin and grow from there. The community welcomes your contributions! 🚀