1
use tokio::net::{TcpListener, TcpStream};
2
use rustls::{ServerConfig, ClientConfig, Certificate, PrivateKey};
3
use std::sync::Arc;
4
use futures::future::try_join;
5
use bytes::{Bytes, BytesMut};
6
use http::{Request, Response, StatusCode};
7

8
pub struct SidecarProxy {
9
    inbound_config: Arc<ServerConfig>,
10
    outbound_config: Arc<ClientConfig>,
11
    policy_engine: Arc<PolicyEngine>,
12
    metrics: Arc<Metrics>,
13
}
14

15
impl SidecarProxy {
16
    pub async fn new(config: ProxyConfig) -> Result<Self> {
17
        // Build TLS configurations with automatic certificate rotation
18
        let inbound_config = Self::build_server_config(&config).await?;
19
        let outbound_config = Self::build_client_config(&config).await?;
20

21
        Ok(Self {
22
            inbound_config: Arc::new(inbound_config),
23
            outbound_config: Arc::new(outbound_config),
24
            policy_engine: Arc::new(PolicyEngine::new(config.policies)),
25
            metrics: Arc::new(Metrics::new()),
26
        })
27
    }
28

29
    pub async fn run(&self, addr: SocketAddr) -> Result<()> {
30
        let listener = TcpListener::bind(addr).await?;
31
        println!("Sidecar proxy listening on {}", addr);
32

33
        loop {
34
            let (inbound, peer_addr) = listener.accept().await?;
35
            let proxy = self.clone();
36

37
            tokio::spawn(async move {
38
                if let Err(e) = proxy.handle_connection(inbound, peer_addr).await {
39
                    eprintln!("Connection error: {}", e);
40
                }
41
            });
42
        }
43
    }
44

45
    async fn handle_connection(
46
        &self,
47
        inbound: TcpStream,
48
        peer_addr: SocketAddr,
49
    ) -> Result<()> {
50
        let start = Instant::now();
51

52
        // Establish mTLS connection
53
        let tls_stream = self.establish_tls(inbound).await?;
54

55
        // Extract client identity from certificate
56
        let client_identity = self.extract_identity(&tls_stream)?;
57

58
        // Parse HTTP request
59
        let request = self.parse_request(&mut tls_stream).await?;
60

61
        // Apply security policies
62
        let decision = self.policy_engine.evaluate(
63
            &client_identity,
64
            &request,
65
            &peer_addr,
66
        ).await?;
67

68
        if !decision.allowed {
69
            self.send_forbidden_response(&mut tls_stream).await?;
70
            self.metrics.record_blocked_request(&client_identity);
71
            return Ok(());
72
        }
73

74
        // Forward to upstream service
75
        let response = self.forward_request(request, &client_identity).await?;
76

77
        // Send response back
78
        self.send_response(&mut tls_stream, response).await?;
79

80
        // Record metrics
81
        let duration = start.elapsed();
82
        self.metrics.record_request(duration, &client_identity);
83

84
        Ok(())
85
    }
86

87
    async fn forward_request(
88
        &self,
89
        mut request: Request<Bytes>,
90
        client_identity: &Identity,
91
    ) -> Result<Response<Bytes>> {
92
        // Add security headers
93
        request.headers_mut().insert(
94
            "X-Client-Identity",
95
            client_identity.to_string().parse()?,
96
        );
97

98
        // Establish mTLS connection to upstream
99
        let upstream = self.connect_upstream(&request).await?;
100

101
        // Forward request and get response
102
        let response = self.send_http_request(upstream, request).await?;
103

104
        Ok(response)
105
    }
106
}
107

108
// Zero-copy HTTP parser for performance
109
pub struct HttpParser {
110
    buffer: BytesMut,
111
    headers_complete: bool,
112
    content_length: Option<usize>,
113
}
114

115
impl HttpParser {
116
    pub fn new() -> Self {
117
        Self {
118
            buffer: BytesMut::with_capacity(8192),
119
            headers_complete: false,
120
            content_length: None,
121
        }
122
    }
123

124
    pub async fn parse_request<R: AsyncRead + Unpin>(
125
        &mut self,
126
        reader: &mut R,
127
    ) -> Result<Request<Bytes>> {
128
        loop {
129
            // Read data into buffer
130
            let n = reader.read_buf(&mut self.buffer).await?;
131
            if n == 0 {
132
                return Err(Error::ConnectionClosed);
133
            }
134

135
            // Try to parse headers
136
            if !self.headers_complete {
137
                if let Some(headers_end) = self.find_headers_end() {
138
                    self.headers_complete = true;
139
                    self.parse_headers(headers_end)?;
140
                }
141
            }
142

143
            // Check if we have complete message
144
            if self.is_complete() {
145
                return self.build_request();
146
            }
147
        }
148
    }
149
}

1
use x509_cert::{Certificate as X509Certificate, TbsCertificate};
2
use ring::{rand, signature};
3
use std::time::{Duration, SystemTime};
4

5
pub struct CertificateManager {
6
    ca_cert: X509Certificate,
7
    ca_key: signature::EcdsaKeyPair,
8
    store: Arc<RwLock<CertificateStore>>,
9
    rotation_interval: Duration,
10
}
11

12
impl CertificateManager {
13
    pub async fn new(config: CertConfig) -> Result<Self> {
14
        let ca_cert = Self::load_ca_cert(&config.ca_cert_path)?;
15
        let ca_key = Self::load_ca_key(&config.ca_key_path)?;
16

17
        let manager = Self {
18
            ca_cert,
19
            ca_key,
20
            store: Arc::new(RwLock::new(CertificateStore::new())),
21
            rotation_interval: Duration::from_secs(config.rotation_interval_secs),
22
        };
23

24
        // Start automatic rotation
25
        manager.start_rotation_task();
26

27
        Ok(manager)
28
    }
29

30
    pub async fn issue_certificate(
31
        &self,
32
        service_name: &str,
33
        san: Vec<String>,
34
    ) -> Result<(Certificate, PrivateKey)> {
35
        // Generate key pair
36
        let rng = rand::SystemRandom::new();
37
        let key_pair = signature::EcdsaKeyPair::generate_pkcs8(
38
            &signature::ECDSA_P256_SHA256_ASN1_SIGNING,
39
            &rng,
40
        )?;
41

42
        // Build certificate
43
        let cert = self.build_certificate(service_name, san, &key_pair)?;
44

45
        // Sign with CA
46
        let signed_cert = self.sign_certificate(cert)?;
47

48
        // Store in cache
49
        self.store.write().await.insert(
50
            service_name.to_string(),
51
            CertificateEntry {
52
                certificate: signed_cert.clone(),
53
                private_key: key_pair,
54
                expires_at: SystemTime::now() + Duration::from_secs(86400), // 24 hours
55
            },
56
        );
57

58
        Ok((signed_cert, key_pair))
59
    }
60

61
    fn start_rotation_task(&self) {
62
        let store = self.store.clone();
63
        let manager = self.clone();
64

65
        tokio::spawn(async move {
66
            let mut interval = tokio::time::interval(Duration::from_secs(60));
67

68
            loop {
69
                interval.tick().await;
70

71
                // Check for certificates near expiry
72
                let now = SystemTime::now();
73
                let mut expired = Vec::new();
74

75
                {
76
                    let store = store.read().await;
77
                    for (service, entry) in store.iter() {
78
                        if entry.expires_at <= now + Duration::from_secs(3600) {
79
                            expired.push(service.clone());
80
                        }
81
                    }
82
                }
83

84
                // Rotate expired certificates
85
                for service in expired {
86
                    if let Err(e) = manager.rotate_certificate(&service).await {
87
                        eprintln!("Failed to rotate certificate for {}: {}", service, e);
88
                    }
89
                }
90
            }
91
        });
92
    }
93

94
    async fn rotate_certificate(&self, service_name: &str) -> Result<()> {
95
        // Generate new certificate
96
        let (new_cert, new_key) = self.issue_certificate(service_name, vec![]).await?;
97

98
        // Atomic rotation with grace period
99
        let grace_period = Duration::from_secs(300); // 5 minutes
100

101
        // Update store with both old and new certificates
102
        self.store.write().await.set_rotation(
103
            service_name,
104
            new_cert,
105
            new_key,
106
            grace_period,
107
        );
108

109
        // Notify services about new certificate
110
        self.notify_certificate_rotation(service_name).await?;
111

112
        Ok(())
113
    }
114
}

1
use serde::{Deserialize, Serialize};
2
use rego::{Rego, Value};
3
use dashmap::DashMap;
4

5
#[derive(Debug, Clone, Serialize, Deserialize)]
6
pub struct Policy {
7
    id: String,
8
    rules: Vec<Rule>,
9
    priority: i32,
10
}
11

12
#[derive(Debug, Clone, Serialize, Deserialize)]
13
pub struct Rule {
14
    source: Matcher,
15
    destination: Matcher,
16
    operation: OperationMatcher,
17
    action: Action,
18
}
19

20
pub struct PolicyEngine {
21
    policies: Arc<RwLock<Vec<Policy>>>,
22
    rego_engine: Option<Rego>,
23
    cache: Arc<DashMap<PolicyKey, Decision>>,
24
    metrics: Arc<PolicyMetrics>,
25
}
26

27
impl PolicyEngine {
28
    pub async fn evaluate(
29
        &self,
30
        identity: &Identity,
31
        request: &Request<Bytes>,
32
        source_addr: &SocketAddr,
33
    ) -> Result<Decision> {
34
        let start = Instant::now();
35

36
        // Build policy evaluation context
37
        let context = PolicyContext {
38
            source: SourceContext {
39
                identity: identity.clone(),
40
                address: *source_addr,
41
                attributes: self.extract_source_attributes(identity),
42
            },
43
            destination: DestinationContext {
44
                service: self.extract_service_name(request),
45
                namespace: self.extract_namespace(request),
46
                path: request.uri().path().to_string(),
47
                method: request.method().clone(),
48
            },
49
            request: RequestContext {
50
                headers: self.extract_headers(request),
51
                time: SystemTime::now(),
52
            },
53
        };
54

55
        // Check cache
56
        let cache_key = self.build_cache_key(&context);
57
        if let Some(cached) = self.cache.get(&cache_key) {
58
            self.metrics.record_cache_hit();
59
            return Ok(cached.clone());
60
        }
61

62
        // Evaluate policies
63
        let decision = if let Some(rego) = &self.rego_engine {
64
            // Use OPA/Rego for complex policies
65
            self.evaluate_rego(&context).await?
66
        } else {
67
            // Use built-in rule engine
68
            self.evaluate_rules(&context).await?
69
        };
70

71
        // Cache decision
72
        self.cache.insert(cache_key, decision.clone());
73

74
        // Record metrics
75
        let duration = start.elapsed();
76
        self.metrics.record_evaluation(duration, &decision);
77

78
        Ok(decision)
79
    }
80

81
    async fn evaluate_rules(&self, context: &PolicyContext) -> Result<Decision> {
82
        let policies = self.policies.read().await;
83

84
        // Sort by priority
85
        let mut sorted_policies = policies.clone();
86
        sorted_policies.sort_by_key(|p| -p.priority);
87

88
        for policy in &sorted_policies {
89
            for rule in &policy.rules {
90
                if self.matches_rule(rule, context)? {
91
                    return Ok(Decision {
92
                        allowed: rule.action == Action::Allow,
93
                        reason: format!("Matched policy: {}", policy.id),
94
                        matched_rule: Some(rule.clone()),
95
                    });
96
                }
97
            }
98
        }
99

100
        // Default deny
101
        Ok(Decision {
102
            allowed: false,
103
            reason: "No matching policy found".to_string(),
104
            matched_rule: None,
105
        })
106
    }
107

108
    fn matches_rule(&self, rule: &Rule, context: &PolicyContext) -> Result<bool> {
109
        // Match source
110
        if !self.matches_source(&rule.source, &context.source)? {
111
            return Ok(false);
112
        }
113

114
        // Match destination
115
        if !self.matches_destination(&rule.destination, &context.destination)? {
116
            return Ok(false);
117
        }
118

119
        // Match operation
120
        if !self.matches_operation(&rule.operation, &context.destination)? {
121
            return Ok(false);
122
        }
123

124
        Ok(true)
125
    }
126
}
127

128
// Rate limiting and circuit breaking
129
pub struct TrafficController {
130
    rate_limiters: Arc<DashMap<String, RateLimiter>>,
131
    circuit_breakers: Arc<DashMap<String, CircuitBreaker>>,
132
}
133

134
impl TrafficController {
135
    pub async fn check_rate_limit(
136
        &self,
137
        identity: &Identity,
138
        service: &str,
139
    ) -> Result<bool> {
140
        let key = format!("{}:{}", identity.name, service);
141

142
        let limiter = self.rate_limiters
143
            .entry(key.clone())
144
            .or_insert_with(|| RateLimiter::new(
145
                100, // 100 requests
146
                Duration::from_secs(60), // per minute
147
            ));
148

149
        Ok(limiter.check())
150
    }
151

152
    pub async fn check_circuit_breaker(
153
        &self,
154
        service: &str,
155
    ) -> Result<CircuitState> {
156
        let breaker = self.circuit_breakers
157
            .entry(service.to_string())
158
            .or_insert_with(|| CircuitBreaker::new(CircuitBreakerConfig {
159
                failure_threshold: 5,
160
                success_threshold: 3,
161
                timeout: Duration::from_secs(30),
162
            }));
163

164
        Ok(breaker.state())
165
    }
166
}

1
use aya::{Bpf, maps::{HashMap as BpfHashMap, PerfEventArray}, programs::{Xdp, XdpFlags}};
2
use aya_bpf::bindings::xdp_action;
3
use std::convert::TryInto;
4

5
pub struct EbpfAccelerator {
6
    bpf: Bpf,
7
    policy_map: BpfHashMap<PolicyKey, PolicyAction>,
8
    events: PerfEventArray<ServiceMeshEvent>,
9
}
10

11
impl EbpfAccelerator {
12
    pub async fn new() -> Result<Self> {
13
        // Load eBPF program
14
        let mut bpf = Bpf::load(include_bytes!(
15
            concat!(env!("OUT_DIR"), "/service_mesh.bpf.o")
16
        ))?;
17

18
        // Attach XDP program
19
        let program: &mut Xdp = bpf.program_mut("service_mesh_filter").unwrap().try_into()?;
20
        program.load()?;
21
        program.attach("eth0", XdpFlags::default())?;
22

23
        // Get maps
24
        let policy_map = BpfHashMap::try_from(bpf.map_mut("POLICY_MAP")?)?;
25
        let events = PerfEventArray::try_from(bpf.map_mut("EVENTS")?)?;
26

27
        Ok(Self {
28
            bpf,
29
            policy_map,
30
            events,
31
        })
32
    }
33

34
    pub async fn update_policy(
35
        &mut self,
36
        source: IpAddr,
37
        dest: IpAddr,
38
        action: PolicyAction,
39
    ) -> Result<()> {
40
        let key = PolicyKey { source, dest };
41
        self.policy_map.insert(&key, &action, 0)?;
42
        Ok(())
43
    }
44

45
    pub async fn start_event_processor(&mut self) -> Result<()> {
46
        let mut perf_array = self.events.clone();
47

48
        tokio::spawn(async move {
49
            let mut buffers = vec![BytesMut::with_capacity(1024); 10];
50

51
            loop {
52
                let events = perf_array.read_events(&mut buffers).unwrap();
53

54
                for buf in &buffers[..events] {
55
                    let event: ServiceMeshEvent = unsafe {
56
                        ptr::read_unaligned(buf.as_ptr() as *const _)
57
                    };
58

59
                    // Process event
60
                    match event.event_type {
61
                        EventType::ConnectionAllowed => {
62
                            // Log allowed connection
63
                        }
64
                        EventType::ConnectionBlocked => {
65
                            // Alert on blocked connection
66
                        }
67
                        EventType::RateLimitExceeded => {
68
                            // Handle rate limit
69
                        }
70
                    }
71
                }
72
            }
73
        });
74

75
        Ok(())
76
    }
77
}
78

79
// eBPF program (compiled separately)
80
#[repr(C)]
81
struct PolicyKey {
82
    source: u32,
83
    dest: u32,
84
}
85

86
#[repr(C)]
87
struct PolicyAction {
88
    allow: u8,
89
    rate_limit: u32,
90
}
91

92
// The actual eBPF code would be in a separate .c file

1
use opentelemetry::{
2
    trace::{Tracer, Span, SpanKind},
3
    propagation::{Extractor, Injector},
4
    Context,
5
};
6
use prometheus::{Counter, Histogram, Registry};
7

8
pub struct ServiceMeshTelemetry {
9
    tracer: Box<dyn Tracer>,
10
    metrics: ServiceMeshMetrics,
11
    log_aggregator: LogAggregator,
12
}
13

14
pub struct ServiceMeshMetrics {
15
    request_count: Counter,
16
    request_duration: Histogram,
17
    connection_count: Counter,
18
    policy_decisions: Counter,
19
    certificate_rotations: Counter,
20
}
21

22
impl ServiceMeshTelemetry {
23
    pub fn new() -> Result<Self> {
24
        // Initialize OpenTelemetry
25
        let tracer = opentelemetry_jaeger::new_pipeline()
26
            .with_service_name("service-mesh")
27
            .install_batch(opentelemetry::runtime::Tokio)?;
28

29
        // Initialize Prometheus metrics
30
        let registry = Registry::new();
31
        let metrics = ServiceMeshMetrics {
32
            request_count: Counter::new("mesh_requests_total", "Total requests")?,
33
            request_duration: Histogram::with_opts(
34
                HistogramOpts::new("mesh_request_duration_seconds", "Request duration")
35
                    .buckets(vec![0.001, 0.005, 0.01, 0.05, 0.1, 0.5, 1.0]),
36
            )?,
37
            connection_count: Counter::new("mesh_connections_total", "Total connections")?,
38
            policy_decisions: Counter::new("mesh_policy_decisions_total", "Policy decisions")?,
39
            certificate_rotations: Counter::new("mesh_cert_rotations_total", "Certificate rotations")?,
40
        };
41

42
        Ok(Self {
43
            tracer: Box::new(tracer),
44
            metrics,
45
            log_aggregator: LogAggregator::new(),
46
        })
47
    }
48

49
    pub fn trace_request<F, R>(&self, operation: &str, f: F) -> R
50
    where
51
        F: FnOnce(&mut Span) -> R,
52
    {
53
        let mut span = self.tracer
54
            .span_builder(operation)
55
            .with_kind(SpanKind::Server)
56
            .start(&self.tracer);
57

58
        let result = f(&mut span);
59

60
        span.end();
61
        result
62
    }
63

64
    pub fn extract_trace_context(&self, headers: &HeaderMap) -> Context {
65
        let extractor = HttpHeaderExtractor(headers);
66
        opentelemetry::global::get_text_map_propagator(|propagator| {
67
            propagator.extract(&extractor)
68
        })
69
    }
70

71
    pub fn inject_trace_context(&self, context: &Context, headers: &mut HeaderMap) {
72
        let mut injector = HttpHeaderInjector(headers);
73
        opentelemetry::global::get_text_map_propagator(|propagator| {
74
            propagator.inject_context(context, &mut injector)
75
        })
76
    }
77
}
78

79
// Distributed tracing for service mesh
80
pub struct DistributedTracer {
81
    spans: Arc<RwLock<HashMap<SpanId, SpanData>>>,
82
    exporter: JaegerExporter,
83
}
84

85
impl DistributedTracer {
86
    pub async fn trace_service_call(
87
        &self,
88
        source: &str,
89
        destination: &str,
90
        operation: &str,
91
    ) -> SpanHandle {
92
        let span_id = SpanId::generate();
93
        let trace_id = TraceId::generate();
94

95
        let span_data = SpanData {
96
            span_id,
97
            trace_id,
98
            parent_span_id: None,
99
            operation: operation.to_string(),
100
            service_name: source.to_string(),
101
            start_time: SystemTime::now(),
102
            tags: HashMap::new(),
103
            logs: Vec::new(),
104
        };
105

106
        self.spans.write().await.insert(span_id, span_data);
107

108
        SpanHandle {
109
            span_id,
110
            tracer: self.clone(),
111
        }
112
    }
113
}

1
use k8s_openapi::api::core::v1::{Service, Endpoints};
2
use kube::{Api, Client, runtime::watcher};
3
use tower::load::p2c::Balance;
4
use tower::ServiceExt;
5

6
pub struct ServiceDiscovery {
7
    k8s_client: Client,
8
    consul_client: Option<ConsulClient>,
9
    cache: Arc<RwLock<ServiceRegistry>>,
10
    watchers: Vec<JoinHandle<()>>,
11
}
12

13
impl ServiceDiscovery {
14
    pub async fn new(config: DiscoveryConfig) -> Result<Self> {
15
        let k8s_client = Client::try_default().await?;
16
        let consul_client = config.consul_endpoint
17
            .map(|endpoint| ConsulClient::new(endpoint));
18

19
        let mut discovery = Self {
20
            k8s_client,
21
            consul_client,
22
            cache: Arc::new(RwLock::new(ServiceRegistry::new())),
23
            watchers: Vec::new(),
24
        };
25

26
        // Start watchers
27
        discovery.start_k8s_watcher().await?;
28
        if discovery.consul_client.is_some() {
29
            discovery.start_consul_watcher().await?;
30
        }
31

32
        Ok(discovery)
33
    }
34

35
    async fn start_k8s_watcher(&mut self) -> Result<()> {
36
        let api: Api<Service> = Api::all(self.k8s_client.clone());
37
        let cache = self.cache.clone();
38

39
        let watcher = tokio::spawn(async move {
40
            let mut stream = watcher(api, ListParams::default()).boxed();
41

42
            while let Some(event) = stream.next().await {
43
                match event {
44
                    Ok(Event::Applied(service)) => {
45
                        let service_info = ServiceInfo::from_k8s_service(&service);
46
                        cache.write().await.add_service(service_info);
47
                    }
48
                    Ok(Event::Deleted(service)) => {
49
                        let name = service.metadata.name.unwrap_or_default();
50
                        cache.write().await.remove_service(&name);
51
                    }
52
                    Err(e) => eprintln!("Watcher error: {}", e),
53
                    _ => {}
54
                }
55
            }
56
        });
57

58
        self.watchers.push(watcher);
59
        Ok(())
60
    }
61

62
    pub async fn resolve_service(&self, name: &str) -> Result<Vec<Endpoint>> {
63
        let registry = self.cache.read().await;
64

65
        if let Some(service) = registry.get_service(name) {
66
            Ok(service.endpoints.clone())
67
        } else {
68
            Err(Error::ServiceNotFound(name.to_string()))
69
        }
70
    }
71

72
    pub async fn create_load_balancer(
73
        &self,
74
        service_name: &str,
75
        strategy: LoadBalancerStrategy,
76
    ) -> Result<Box<dyn LoadBalancer>> {
77
        let endpoints = self.resolve_service(service_name).await?;
78

79
        match strategy {
80
            LoadBalancerStrategy::RoundRobin => {
81
                Ok(Box::new(RoundRobinBalancer::new(endpoints)))
82
            }
83
            LoadBalancerStrategy::LeastConnection => {
84
                Ok(Box::new(LeastConnectionBalancer::new(endpoints)))
85
            }
86
            LoadBalancerStrategy::P2C => {
87
                Ok(Box::new(P2CBalancer::new(endpoints)))
88
            }
89
            LoadBalancerStrategy::Consistent => {
90
                Ok(Box::new(ConsistentHashBalancer::new(endpoints)))
91
            }
92
        }
93
    }
94
}
95

96
// Power of Two Choices load balancer
97
pub struct P2CBalancer {
98
    endpoints: Vec<WeightedEndpoint>,
99
    rng: StdRng,
100
}
101

102
impl P2CBalancer {
103
    pub fn pick_endpoint(&mut self) -> &Endpoint {
104
        let n = self.endpoints.len();
105
        if n == 1 {
106
            return &self.endpoints[0].endpoint;
107
        }
108

109
        // Pick two random endpoints
110
        let i = self.rng.gen_range(0..n);
111
        let j = loop {
112
            let x = self.rng.gen_range(0..n);
113
            if x != i {
114
                break x;
115
            }
116
        };
117

118
        // Choose the one with fewer active connections
119
        if self.endpoints[i].active_connections < self.endpoints[j].active_connections {
120
            &self.endpoints[i].endpoint
121
        } else {
122
            &self.endpoints[j].endpoint
123
        }
124
    }
125
}