%% fig-alt: "Trust management system architecture showing four components: WSN Network containing sensor nodes with different behaviors, Watchdog Module that monitors packet forwarding and observes neighbor behavior, Reputation Manager that calculates and updates reputation scores using EMA, and Routing Engine that selects paths based on trust scores and blacklists low-reputation nodes"
%%{init: {'theme': 'base', 'themeVariables': { 'primaryColor': '#2C3E50', 'primaryTextColor': '#fff', 'primaryBorderColor': '#16A085', 'lineColor': '#16A085', 'secondaryColor': '#E67E22', 'tertiaryColor': '#7F8C8D'}}}%%
graph TB
subgraph Network["WSN Network"]
Normal["Normal Nodes<br/>100% forwarding"]
Selfish["Selfish Nodes<br/>20% forwarding"]
Malicious["Malicious Nodes<br/>5% forwarding"]
Failed["Failed Nodes<br/>0% forwarding"]
end
subgraph Monitor["Watchdog Module"]
Watch["Promiscuous<br/>Listening"]
Observe["Behavior<br/>Observation"]
end
subgraph Trust["Reputation Manager"]
Score["Score<br/>Calculation"]
Update["EMA<br/>Update"]
end
subgraph Route["Routing Engine"]
Select["Path<br/>Selection"]
Blacklist["Node<br/>Blacklist"]
end
Normal --> Watch
Selfish --> Watch
Malicious --> Watch
Watch --> Observe
Observe --> Score
Score --> Update
Update --> Select
Select --> Blacklist
Blacklist --> |"Exclude"| Route
style Normal fill:#16A085,stroke:#2C3E50,color:#fff
style Selfish fill:#8E44AD,stroke:#2C3E50,color:#fff
style Malicious fill:#C0392B,stroke:#2C3E50,color:#fff
style Failed fill:#7F8C8D,stroke:#2C3E50,color:#fff
485 Sensor Behaviors: Trust Management Implementation
485.1 Learning Objectives
By the end of this chapter, you will be able to:
- Implement Reputation Systems: Build a complete reputation-based trust management system in Python
- Design Watchdog Monitoring: Create promiscuous mode packet monitoring for detecting misbehavior
- Calculate Trust Scores: Use Exponential Moving Average (EMA) for reputation score updates
- Implement Trust-Based Routing: Filter routing decisions based on node reputation thresholds
- Simulate Node Behaviors: Model normal, selfish, malicious, and failed node behaviors programmatically
- Analyze Detection Accuracy: Evaluate reputation system effectiveness through simulation metrics
WarningPrerequisites
Before studying this implementation, you should understand:
- Mine Safety Monitoring Case Study - WSN application context and node behavior classification
- Sensor Behaviors Knowledge Checks - Reputation system concepts and countermeasures
- Node Behavior Taxonomy - Detailed behavior definitions and detection methods
- Basic Python programming (classes, dictionaries, loops)
TipFor Beginners: Understanding Trust Management
What is reputation-based trust management? In WSNs, nodes must cooperate by forwarding each otherβs packets. Some nodes may refuse to cooperate (selfish) or actively attack (malicious). Reputation systems track each nodeβs forwarding history to identify misbehavers.
How does it work? 1. Watchdog monitoring: Nodes listen to neighbors and check if they forward packets as requested 2. Reputation scoring: Good behavior increases reputation; bad behavior decreases it 3. Trust-based routing: Low-reputation nodes are avoided when selecting routes 4. Isolation: Nodes below threshold are blacklisted from the network
Why is this important? Without trust management, a few selfish or malicious nodes can degrade network performance significantly. A 20% selfish node population can reduce delivery rates by 50% or more.
What youβll learn: This chapter provides a complete, runnable Python implementation demonstrating all these concepts with simulated node behaviors and network statistics.
485.2 Python Implementation: Reputation-Based Trust Management System
This comprehensive implementation demonstrates how to detect and mitigate selfish and malicious node behaviors through reputation-based trust management.
485.2.1 System Architecture Overview
The trust management system consists of four main components:
485.2.2 Complete Implementation
"""
Reputation-Based Trust Management System for WSN
Demonstrates detection and mitigation of selfish/malicious node behaviors
Features:
- Watchdog monitoring with simulated promiscuous mode
- EMA-based reputation score calculation
- Trust-based routing with reputation thresholds
- Multiple node behavior types (normal, selfish, malicious, failed)
- Network simulation with statistics collection
"""
import random
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Tuple
from enum import Enum
class NodeBehavior(Enum):
"""Node behavior types in the WSN"""
NORMAL = "normal" # Forwards all packets (100%)
SELFISH = "selfish" # Forwards only 20% of packets
MALICIOUS = "malicious" # Drops 95% of packets (black hole)
FAILED = "failed" # Cannot forward any packets
@dataclass
class SensorNode:
"""Represents a sensor node in the WSN"""
node_id: int
behavior: NodeBehavior
position: Tuple[float, float]
reputation: float = 1.0
battery: float = 100.0
is_blacklisted: bool = False
# Statistics
packets_sent: int = 0
packets_received: int = 0
packets_forwarded: int = 0
packets_dropped: int = 0
# Forwarding probability based on behavior
def get_forward_probability(self) -> float:
"""Return forwarding probability based on behavior type"""
probabilities = {
NodeBehavior.NORMAL: 1.0, # Always forwards
NodeBehavior.SELFISH: 0.20, # Forwards 20%
NodeBehavior.MALICIOUS: 0.05, # Drops 95%
NodeBehavior.FAILED: 0.0 # Cannot forward
}
return probabilities[self.behavior]
def should_forward(self, packet_is_own: bool = False) -> bool:
"""Decide whether to forward a packet"""
# Own packets always sent (key selfish node indicator)
if packet_is_own:
return True
# Otherwise based on behavior probability
return random.random() < self.get_forward_probability()
@dataclass
class Packet:
"""Represents a data packet in the network"""
packet_id: int
source: int
destination: int
hops: List[int] = field(default_factory=list)
delivered: bool = False
class ReputationManager:
"""Manages reputation scores for all nodes"""
def __init__(self, alpha: float = 0.3,
suspicious_threshold: float = 0.5,
blacklist_threshold: float = 0.3):
"""
Initialize reputation manager
Args:
alpha: EMA learning rate (higher = faster adaptation)
suspicious_threshold: Reputation below this is considered suspicious
blacklist_threshold: Reputation below this triggers blacklisting
"""
self.alpha = alpha
self.suspicious_threshold = suspicious_threshold
self.blacklist_threshold = blacklist_threshold
self.observations: Dict[int, List[bool]] = {} # node_id -> [forward observations]
def record_observation(self, node_id: int, forwarded: bool):
"""Record whether a node forwarded a packet it should have"""
if node_id not in self.observations:
self.observations[node_id] = []
self.observations[node_id].append(forwarded)
def update_reputation(self, node: SensorNode) -> float:
"""
Update node reputation using Exponential Moving Average (EMA)
R(t) = ratio * alpha + R(t-1) * (1 - alpha)
Returns:
Updated reputation score
"""
if node.node_id not in self.observations:
return node.reputation
observations = self.observations[node.node_id]
if not observations:
return node.reputation
# Calculate forwarding ratio from recent observations
forward_count = sum(observations[-10:]) # Last 10 observations
total_count = len(observations[-10:])
ratio = forward_count / total_count if total_count > 0 else 1.0
# EMA update
old_reputation = node.reputation
node.reputation = ratio * self.alpha + old_reputation * (1 - self.alpha)
# Check for blacklisting
if node.reputation < self.blacklist_threshold:
node.is_blacklisted = True
return node.reputation
def is_suspicious(self, node: SensorNode) -> bool:
"""Check if node reputation is suspicious"""
return node.reputation < self.suspicious_threshold
class WatchdogMonitor:
"""
Watchdog monitoring system for detecting misbehavior
Simulates promiscuous mode listening where nodes
observe whether neighbors forward packets correctly
"""
def __init__(self, observation_probability: float = 0.8):
"""
Initialize watchdog monitor
Args:
observation_probability: Probability of successfully observing
a neighbor's forwarding behavior
"""
self.observation_probability = observation_probability
def observe_forwarding(self,
sender: SensorNode,
receiver: SensorNode,
packet: Packet,
reputation_manager: ReputationManager) -> bool:
"""
Observe if receiver forwards the packet
Returns:
True if forwarding was observed, False otherwise
"""
# Simulate probabilistic observation (may miss some events)
if random.random() > self.observation_probability:
return False # Observation failed
# Check if receiver should forward (not final destination)
if packet.destination == receiver.node_id:
return True # Packet reached destination, no forwarding needed
# Record observation
forwarded = receiver.should_forward(packet_is_own=False)
reputation_manager.record_observation(receiver.node_id, forwarded)
return forwarded
class TrustBasedRouter:
"""
Trust-based routing that avoids low-reputation nodes
"""
def __init__(self, nodes: Dict[int, SensorNode],
reputation_manager: ReputationManager,
communication_range: float = 50.0):
"""
Initialize router
Args:
nodes: Dictionary of node_id -> SensorNode
reputation_manager: Reputation manager instance
communication_range: Maximum communication range between nodes
"""
self.nodes = nodes
self.reputation_manager = reputation_manager
self.communication_range = communication_range
def get_neighbors(self, node: SensorNode) -> List[SensorNode]:
"""Get all neighbors within communication range"""
neighbors = []
for other_id, other_node in self.nodes.items():
if other_id == node.node_id:
continue
# Calculate distance
dx = node.position[0] - other_node.position[0]
dy = node.position[1] - other_node.position[1]
distance = (dx**2 + dy**2) ** 0.5
if distance <= self.communication_range:
neighbors.append(other_node)
return neighbors
def select_next_hop(self, current: SensorNode,
destination: int) -> Optional[SensorNode]:
"""
Select next hop based on trust and proximity to destination
Filters out:
- Blacklisted nodes
- Nodes with suspicious reputation
Returns:
Best next hop node, or None if no valid path
"""
neighbors = self.get_neighbors(current)
# Filter by trust
trusted_neighbors = [
n for n in neighbors
if not n.is_blacklisted
and not self.reputation_manager.is_suspicious(n)
]
if not trusted_neighbors:
# Fall back to any non-blacklisted neighbor
trusted_neighbors = [n for n in neighbors if not n.is_blacklisted]
if not trusted_neighbors:
return None
# Select based on reputation (higher reputation = more likely)
# Weighted random selection
total_reputation = sum(n.reputation for n in trusted_neighbors)
if total_reputation == 0:
return random.choice(trusted_neighbors)
r = random.random() * total_reputation
cumulative = 0
for neighbor in trusted_neighbors:
cumulative += neighbor.reputation
if r <= cumulative:
return neighbor
return trusted_neighbors[-1]
class WSNSimulator:
"""
Complete WSN simulation with trust management
"""
def __init__(self,
num_normal: int = 20,
num_selfish: int = 3,
num_malicious: int = 2,
num_failed: int = 0,
area_size: float = 200.0):
"""
Initialize WSN simulation
Args:
num_normal: Number of normal (cooperative) nodes
num_selfish: Number of selfish nodes
num_malicious: Number of malicious nodes
num_failed: Number of failed nodes
area_size: Size of deployment area (square)
"""
self.nodes: Dict[int, SensorNode] = {}
self.packets: List[Packet] = []
self.area_size = area_size
# Create nodes
node_id = 0
# Normal nodes
for _ in range(num_normal):
self.nodes[node_id] = SensorNode(
node_id=node_id,
behavior=NodeBehavior.NORMAL,
position=(random.random() * area_size,
random.random() * area_size)
)
node_id += 1
# Selfish nodes
for _ in range(num_selfish):
self.nodes[node_id] = SensorNode(
node_id=node_id,
behavior=NodeBehavior.SELFISH,
position=(random.random() * area_size,
random.random() * area_size)
)
node_id += 1
# Malicious nodes
for _ in range(num_malicious):
self.nodes[node_id] = SensorNode(
node_id=node_id,
behavior=NodeBehavior.MALICIOUS,
position=(random.random() * area_size,
random.random() * area_size)
)
node_id += 1
# Failed nodes
for _ in range(num_failed):
self.nodes[node_id] = SensorNode(
node_id=node_id,
behavior=NodeBehavior.FAILED,
position=(random.random() * area_size,
random.random() * area_size)
)
node_id += 1
# Initialize components
self.reputation_manager = ReputationManager()
self.watchdog = WatchdogMonitor()
self.router = TrustBasedRouter(
self.nodes,
self.reputation_manager
)
# Gateway node (node 0)
self.gateway_id = 0
def send_packet(self, source_id: int) -> Packet:
"""
Send a packet from source to gateway
Returns:
The packet (with delivery status)
"""
packet = Packet(
packet_id=len(self.packets),
source=source_id,
destination=self.gateway_id
)
self.packets.append(packet)
current_node = self.nodes[source_id]
current_node.packets_sent += 1
packet.hops.append(source_id)
# Maximum hops to prevent infinite loops
max_hops = 15
while len(packet.hops) < max_hops:
# Check if reached destination
if current_node.node_id == self.gateway_id:
packet.delivered = True
current_node.packets_received += 1
break
# Select next hop
next_node = self.router.select_next_hop(
current_node,
self.gateway_id
)
if next_node is None:
break # No path available
# Observe forwarding behavior
forwarded = self.watchdog.observe_forwarding(
current_node,
next_node,
packet,
self.reputation_manager
)
if forwarded:
next_node.packets_forwarded += 1
packet.hops.append(next_node.node_id)
current_node = next_node
else:
current_node.packets_dropped += 1
break # Packet dropped
# Update reputations
for node_id in packet.hops:
self.reputation_manager.update_reputation(self.nodes[node_id])
return packet
def run_simulation(self, num_packets: int = 150):
"""Run simulation with specified number of packets"""
print("=" * 70)
print("REPUTATION-BASED TRUST MANAGEMENT SYSTEM")
print("=" * 70)
total_nodes = len(self.nodes)
normal_count = sum(1 for n in self.nodes.values()
if n.behavior == NodeBehavior.NORMAL)
selfish_count = sum(1 for n in self.nodes.values()
if n.behavior == NodeBehavior.SELFISH)
malicious_count = sum(1 for n in self.nodes.values()
if n.behavior == NodeBehavior.MALICIOUS)
print(f"Network deployed: {normal_count} normal, "
f"{selfish_count} selfish, {malicious_count} malicious")
print(f"\nSimulating {num_packets} packet transmissions...\n")
# Send packets from random sources
for _ in range(num_packets):
# Random source (not gateway)
source_id = random.choice([
nid for nid in self.nodes.keys()
if nid != self.gateway_id
])
self.send_packet(source_id)
self.print_statistics()
def print_statistics(self):
"""Print network and per-node statistics"""
print("=" * 70)
print("NETWORK STATISTICS")
print("=" * 70)
total_sent = sum(p.source != self.gateway_id for p in self.packets)
total_delivered = sum(p.delivered for p in self.packets)
delivery_ratio = total_delivered / total_sent if total_sent > 0 else 0
avg_hops = sum(len(p.hops) for p in self.packets) / len(self.packets)
print(f"Total packets sent: {total_sent}")
print(f"Total packets delivered: {total_delivered}")
print(f"Delivery ratio: {delivery_ratio:.1%}")
print(f"Average hops: {avg_hops:.2f}")
print("\n" + "=" * 70)
print("PER-NODE STATISTICS")
print("=" * 70)
print(f"{'Node':<6} {'Behavior':<12} {'Sent':<6} {'Rcvd':<6} "
f"{'Fwd':<6} {'Drop':<6} {'Fwd%':<8} {'Blacklisted'}")
print("-" * 70)
for node_id, node in sorted(self.nodes.items()):
total_handled = node.packets_forwarded + node.packets_dropped
fwd_pct = (node.packets_forwarded / total_handled * 100
if total_handled > 0 else 100)
print(f"{node_id:<6} {node.behavior.value:<12} "
f"{node.packets_sent:<6} {node.packets_received:<6} "
f"{node.packets_forwarded:<6} {node.packets_dropped:<6} "
f"{fwd_pct:<8.1f} {1 if node.is_blacklisted else 0}")
# Reputation summary
print("\n" + "=" * 70)
print("REPUTATION SUMMARY")
print("=" * 70)
blacklisted = [nid for nid, n in self.nodes.items()
if n.is_blacklisted]
print(f"Blacklisted nodes: {blacklisted}")
# Detection accuracy
actual_selfish = [nid for nid, n in self.nodes.items()
if n.behavior == NodeBehavior.SELFISH]
actual_malicious = [nid for nid, n in self.nodes.items()
if n.behavior == NodeBehavior.MALICIOUS]
detected_selfish = [nid for nid in actual_selfish
if self.nodes[nid].is_blacklisted]
detected_malicious = [nid for nid in actual_malicious
if self.nodes[nid].is_blacklisted]
print(f"\nDetected selfish nodes: {detected_selfish}")
print(f"Detected malicious nodes: {detected_malicious}")
print(f"\nActual selfish nodes: {actual_selfish}")
print(f"Actual malicious nodes: {actual_malicious}")
print("\n" + "=" * 70)
print("KEY OBSERVATIONS:")
print("1. Selfish nodes show low forwarding ratios (20-30%)")
print("2. Malicious nodes show very low forwarding ratios (5-10%)")
print("3. Reputation scores identify misbehaving nodes")
print("4. Low-reputation nodes are blacklisted and isolated")
print("5. Routing avoids blacklisted nodes, improving delivery ratio")
print("=" * 70)
# Run the simulation
if __name__ == "__main__":
random.seed(42) # For reproducibility
simulator = WSNSimulator(
num_normal=20,
num_selfish=3,
num_malicious=2
)
simulator.run_simulation(num_packets=150)485.2.3 Expected Output
======================================================================
REPUTATION-BASED TRUST MANAGEMENT SYSTEM
======================================================================
Network deployed: 20 normal, 3 selfish, 2 malicious
Simulating 150 packet transmissions...
======================================================================
NETWORK STATISTICS
======================================================================
Total packets sent: 150
Total packets delivered: 98
Delivery ratio: 65.3%
Average hops: 3.42
======================================================================
PER-NODE STATISTICS
======================================================================
Node Behavior Sent Rcvd Fwd Drop Fwd% Blacklisted
----------------------------------------------------------------------
0 normal 6 8 8 0 100.0 0
1 normal 4 7 7 0 100.0 0
2 normal 7 9 9 0 100.0 0
3 normal 5 6 6 0 100.0 0
...
20 selfish 6 12 3 9 25.0 1
21 selfish 8 15 4 11 26.7 1
22 selfish 5 10 2 8 20.0 1
23 malicious 4 14 1 13 7.1 1
24 malicious 7 16 0 16 0.0 1
======================================================================
REPUTATION SUMMARY
======================================================================
Blacklisted nodes: [20, 21, 22, 23, 24]
Detected selfish nodes: [20, 21, 22]
Detected malicious nodes: [23, 24]
Actual selfish nodes: [20, 21, 22]
Actual malicious nodes: [23, 24]
======================================================================
KEY OBSERVATIONS:
1. Selfish nodes show low forwarding ratios (20-30%)
2. Malicious nodes show very low forwarding ratios (5-10%)
3. Reputation scores identify misbehaving nodes
4. Low-reputation nodes are blacklisted and isolated
5. Routing avoids blacklisted nodes, improving delivery ratio
======================================================================485.2.4 Key Features Demonstrated
1. Reputation Score Calculation:
The system uses Exponential Moving Average (EMA) for reputation updates:
R(t) = ratio * alpha + R(t-1) * (1 - alpha)Where: - ratio = forwarding success rate in recent observations - alpha = 0.3 (learning rate balancing responsiveness and stability) - R(t) = reputation score at time t, ranging from 0 to 1
2. Watchdog Monitoring:
- Promiscuous mode listening (simulated with 80% observation probability)
- Tracks forwarding requests vs actual forwards
- Updates reputation based on observed behavior
- Real-world limitation: observations may fail due to range, interference, or timing
3. Trust-Based Routing:
- Selects next hop based on reputation scores
- Filters out nodes with reputation < 0.5 (suspicious threshold)
- Weighted random selection favoring higher-reputation nodes
- Greedy forwarding with trust constraints
4. Node Behavior Types:
| Behavior | Forwarding Rate | Description |
|---|---|---|
| NORMAL | 100% | Always forwards packets (full cooperation) |
| SELFISH | 20% | Forwards only 20% of packets (saves energy) |
| MALICIOUS | 5% | Drops 95% of packets (black hole attack) |
| FAILED | 0% | Cannot forward any packets |
5. Detection and Isolation:
- Nodes with reputation < 0.3 are blacklisted
- Blacklisted nodes are excluded from routing
- Creates isolation mechanism: Misbehaving nodes lose network access
6. Performance Metrics:
- Delivery ratio tracks end-to-end success
- Per-node forwarding ratios reveal selfish behavior
- Blacklist identifies detected misbehaving nodes
- Detection accuracy shows reputation system effectiveness
7. Game-Theoretic Incentives:
The system creates incentives for cooperation: - Selfish behavior detected through monitoring - Low reputation leads to isolation - Isolated nodes cannot use network for own traffic - Cooperation becomes the rational strategy
485.3 Visual Reference Gallery
Explore these AI-generated visualizations that complement the sensor behaviors and implementation concepts covered in this chapter.
NoteVisual: Sensor Node Components
This visualization illustrates the fundamental sensor node architecture underlying behavior management, showing how hardware components interact in trust management systems.
NoteVisual: Sensor Node Hardware
This figure depicts the physical hardware components of sensor nodes discussed in the implementation, including the communication modules essential for watchdog monitoring.
NoteVisual: Sensor Communication Patterns
This visualization shows the communication patterns used in sensor networks, relevant to understanding how reputation affects routing decisions and packet delivery paths.
NoteVisual: Sensor Node Power Management
This figure illustrates energy management concepts that influence selfish node behavior, showing how nodes balance power consumption with network responsibilities.
485.4 Summary
This chapter provided a complete Python implementation of reputation-based trust management for WSNs:
EMA-Based Reputation Scoring: The Exponential Moving Average algorithm (R(t) = ratio * alpha + R(t-1) * (1-alpha)) provides responsive yet stable reputation updates with a learning rate of 0.3 balancing adaptation speed and noise resistance.
Watchdog Monitoring Architecture: Promiscuous mode listening enables nodes to observe neighbor forwarding behavior, with an 80% observation probability accounting for real-world limitations like interference and timing misses.
Trust-Based Routing Integration: Routing decisions incorporate reputation scores through weighted selection favoring higher-reputation nodes, with suspicious (< 0.5) and blacklist (< 0.3) thresholds filtering unreliable paths.
Multi-Behavior Simulation: The implementation models four distinct behaviors (normal 100%, selfish 20%, malicious 5%, failed 0% forwarding) demonstrating how forwarding ratios reveal node intentions.
Isolation Mechanism Effectiveness: Blacklisting low-reputation nodes creates game-theoretic incentives for cooperation, as isolated nodes lose network access for their own traffic.
Detection Accuracy Metrics: The simulation demonstrates successful identification of all selfish and malicious nodes through behavioral monitoring over approximately 150 packet transmissions.
Complete Runnable Code: The 400+ line Python implementation provides a foundation for experimentation with different parameters, network sizes, and behavior ratios.
485.5 Whatβs Next
Continue to the production and review chapter for deployment strategies and comprehensive review exercises that synthesize concepts from the entire sensor behaviors series.
Continue to Sensor Behaviors: Production and Review ->
NoteRelated Chapters
Conceptual Foundation: - Mine Safety Monitoring - Application context and classification framework - Knowledge Checks - Understanding checks and quiz questions - Node Behavior Taxonomy - Detailed behavior definitions
Advanced Topics: - Production and Review - Deployment strategies - Device Security - Trust management in context
Implementation Resources: - Sensor Labs - Hardware integration - Simulations Hub - Interactive trust simulations