1use crate::server::ics_support::{number, DeviceSession, ScannerSession};
3use anyhow::{bail, ensure, Context, Result};
4use serde_json::{json, Value};
5use std::{collections::VecDeque, time::Duration};
6use tokio::{
7 io::{AsyncReadExt, AsyncWriteExt},
8 net::TcpStream,
9 time::Instant,
10};
11pub const MAX_FRAME: usize = 255;
12pub const T1: Duration = Duration::from_secs(15);
13pub const T3: Duration = Duration::from_secs(20);
14pub const K: usize = 12;
15pub const W: usize = 8;
16pub fn u_frame(code: u8) -> Vec<u8> {
17 vec![0x68, 4, code, 0, 0, 0]
18}
19fn address(b: &[u8]) -> u32 {
20 b[0] as u32 | ((b[1] as u32) << 8) | ((b[2] as u32) << 16)
21}
22fn asdu(typ: u8, cause: u8, origin: u8, ca: u16, ioa: u32, data: &[u8]) -> Vec<u8> {
23 let mut b = vec![typ, 1, cause, origin];
24 b.extend(ca.to_le_bytes());
25 b.extend(&ioa.to_le_bytes()[..3]);
26 b.extend(data);
27 b
28}
29#[derive(Default)]
30struct Channel {
31 send: u16,
32 recv: u16,
33 acked: u16,
34 pending: VecDeque<(u16, Instant)>,
35 active: bool,
36 buffer: Vec<u8>,
37 test: Option<Instant>,
38}
39impl Channel {
40 fn i(&mut self, asdu: &[u8]) -> Result<Vec<u8>> {
41 ensure!(
42 self.active && asdu.len() <= 249 && self.pending.len() < K,
43 "inactive link or send window exceeded"
44 );
45 let mut b = vec![0x68, (asdu.len() + 4) as u8];
46 b.extend((self.send << 1).to_le_bytes());
47 b.extend((self.recv << 1).to_le_bytes());
48 b.extend(asdu);
49 self.pending.push_back((self.send, Instant::now()));
50 self.send = (self.send + 1) & 32767;
51 Ok(b)
52 }
53 fn s(&self) -> Vec<u8> {
54 let mut b = vec![0x68, 4, 1, 0];
55 b.extend((self.recv << 1).to_le_bytes());
56 b
57 }
58 fn ack(&mut self, ack: u16) -> Result<()> {
59 let distance = ack.wrapping_sub(self.acked) & 32767;
60 ensure!(
61 distance as usize <= self.pending.len(),
62 "invalid acknowledgment sequence"
63 );
64 for _ in 0..distance {
65 self.pending.pop_front();
66 }
67 self.acked = ack;
68 Ok(())
69 }
70 fn receive(&mut self, f: &[u8]) -> Result<(Vec<u8>, Option<Vec<u8>>)> {
71 ensure!(
72 f.len() >= 6 && f[0] == 0x68 && f[1] as usize + 2 == f.len(),
73 "APCI length/version"
74 );
75 if f[2] & 1 == 0 {
76 ensure!(
77 self.active && f.len() >= 15 && f[4] & 1 == 0,
78 "I frame before STARTDT or truncated ASDU"
79 );
80 let sent = u16::from_le_bytes([f[2], f[3]]) >> 1;
81 let ack = u16::from_le_bytes([f[4], f[5]]) >> 1;
82 self.ack(ack)?;
83 ensure!(sent == self.recv, "receive sequence mismatch");
84 self.recv = (self.recv + 1) & 32767;
85 return Ok((vec![], Some(f[6..].to_vec())));
86 }
87 ensure!(f.len() == 6, "control frame has ASDU");
88 if f[2] & 3 == 1 {
89 ensure!(f[2..4] == [1, 0] && f[4] & 1 == 0, "bad S frame");
90 self.ack(u16::from_le_bytes([f[4], f[5]]) >> 1)?;
91 return Ok((vec![], None));
92 }
93 ensure!(f[3..6] == [0; 3], "bad U frame reserved fields");
94 let reply = match f[2] {
95 7 => {
96 self.active = true;
97 u_frame(0x0b)
98 }
99 0x0b => {
100 self.active = true;
101 vec![]
102 }
103 0x13 => {
104 self.active = false;
105 u_frame(0x23)
106 }
107 0x23 => {
108 self.active = false;
109 vec![]
110 }
111 0x43 => u_frame(0x83),
112 0x83 => {
113 ensure!(self.test.is_some(), "unexpected TESTFR confirmation");
114 self.test = None;
115 vec![]
116 }
117 _ => bail!("unknown U frame"),
118 };
119 Ok((reply, None))
120 }
121 async fn read(&mut self, s: &mut TcpStream) -> Result<Vec<u8>> {
123 let mut idle = Instant::now() + T3;
124 loop {
125 if self.buffer.len() >= 2 {
126 ensure!(
127 self.buffer[0] == 0x68 && (4..=253).contains(&self.buffer[1]),
128 "invalid APCI framing"
129 );
130 let n = self.buffer[1] as usize + 2;
131 if self.buffer.len() >= n {
132 return Ok(self.buffer.drain(..n).collect());
133 }
134 }
135 let deadline = self
136 .pending
137 .front()
138 .map(|(_, t)| *t + T1)
139 .unwrap_or(idle)
140 .min(self.test.map(|t| t + T1).unwrap_or(idle));
141 let mut b = [0; 255];
142 match tokio::time::timeout_at(deadline, s.read(&mut b)).await {
143 Ok(Ok(0)) => bail!("IEC 104 peer closed"),
144 Ok(Ok(n)) => {
145 ensure!(
146 self.buffer.len() + n <= MAX_FRAME * 2,
147 "APCI accumulator limit"
148 );
149 self.buffer.extend(&b[..n]);
150 }
151 Ok(Err(e)) => return Err(e.into()),
152 Err(_) => {
153 ensure!(
154 self.pending.is_empty() && self.test.is_none() && self.buffer.is_empty(),
155 "t1 acknowledgment/TESTFR or partial-frame deadline"
156 );
157 s.write_all(&u_frame(0x43)).await?;
158 self.test = Some(Instant::now());
159 idle = Instant::now() + T3;
160 }
161 }
162 }
163 }
164}
165fn decode_telemetry(a: &[u8]) -> Result<Vec<Value>> {
166 ensure!(a.len() >= 6, "ASDU header");
167 let typ = a[0];
168 ensure!(matches!(typ, 1 | 13), "unsupported telemetry ASDU");
169 let count = (a[1] & 127) as usize;
170 ensure!((1..=64).contains(&count), "telemetry count outside bound");
171 let sequential = a[1] & 128 != 0;
172 let ca = u16::from_le_bytes([a[4], a[5]]);
173 let mut pos = 6;
174 let mut start = 0;
175 let mut out = vec![];
176 for i in 0..count {
177 let ioa = if sequential && i > 0 {
178 start + i as u32
179 } else {
180 ensure!(pos + 3 <= a.len(), "IOA truncated");
181 let n = address(&a[pos..]);
182 pos += 3;
183 if i == 0 {
184 start = n;
185 }
186 n
187 };
188 ensure!(ioa <= 0xffffff, "IOA sequence overflow");
189 let (kind, value, quality) = if typ == 1 {
190 ensure!(pos < a.len(), "single point truncated");
191 let p = a[pos];
192 pos += 1;
193 ("binary", json!(p & 1 != 0), p & 0xf0)
194 } else {
195 ensure!(pos + 5 <= a.len(), "float point truncated");
196 let p = f32::from_le_bytes(a[pos..pos + 4].try_into()?);
197 ensure!(p.is_finite(), "nonfinite telemetry");
198 let q = a[pos + 4];
199 pos += 5;
200 ("analog", json!(p), q)
201 };
202 out.push(json!({"kind":kind,"value":value,"quality":quality,"common_address":ca,"ioa":ioa,"cause":a[2]&63}));
203 }
204 ensure!(pos == a.len(), "ASDU trailing data");
205 Ok(out)
206}
207fn point(p: &Value, ca: u16, cause: u8) -> Result<Vec<u8>> {
208 let ioa = number(p, "ioa", 0xffffff)? as u32;
209 let quality = p
210 .get("quality")
211 .map(|v| v.as_u64().context("quality integer"))
212 .transpose()?
213 .unwrap_or(0);
214 ensure!(quality <= 255, "quality byte");
215 match p["kind"].as_str() {
216 Some("binary") => {
217 let v = p["value"].as_bool().context("binary boolean")?;
218 Ok(asdu(
219 1,
220 cause,
221 0,
222 ca,
223 ioa,
224 &[(quality as u8 & 0xf0) | u8::from(v)],
225 ))
226 }
227 Some("analog") => {
228 let n = p["value"].as_f64().context("analog number")?;
229 ensure!(n.is_finite() && (n as f32).is_finite(), "nonfinite analog");
230 let mut b = (n as f32).to_le_bytes().to_vec();
231 b.push(quality as u8);
232 Ok(asdu(13, cause, 0, ca, ioa, &b))
233 }
234 _ => bail!("unsupported point kind"),
235 }
236}
237#[derive(Default)]
238pub struct Device {
239 channel: Channel,
240 request: Vec<u8>,
241}
242#[async_trait::async_trait]
243impl DeviceSession for Device {
244 async fn read(&mut self, s: &mut TcpStream) -> Result<Vec<u8>> {
245 self.channel.read(s).await
246 }
247 fn receive(&mut self, f: &[u8]) -> Result<(Vec<u8>, Option<Value>)> {
248 let (auto, a) = self.channel.receive(f)?;
249 let Some(a) = a else { return Ok((auto, None)) };
250 ensure!(
251 a.len() >= 9 && a[1] == 1,
252 "ASDU expects one non-sequential request"
253 );
254 let ca = u16::from_le_bytes([a[4], a[5]]);
255 let ioa = address(&a[6..]);
256 let cause = a[2] & 63;
257 self.request = a.clone();
258 let mut r = json!({"common_address":ca,"ioa":ioa,"originator":a[3]});
259 match a[0] {
260 100 => {
261 ensure!(
262 a.len() == 10 && ioa == 0 && cause == 6,
263 "invalid interrogation"
264 );
265 r["operation"] = json!("interrogate");
266 r["qualifier"] = json!(a[9]);
267 }
268 102 => {
269 ensure!(a.len() == 9 && cause == 5, "invalid read command");
270 r["operation"] = json!("read");
271 }
272 45 => {
273 ensure!(
274 a.len() == 10 && cause == 6 && a[9] & 2 == 0,
275 "invalid single command"
276 );
277 r["operation"] = json!("command");
278 r["value"] = json!(a[9] & 1 != 0);
279 r["select"] = json!(a[9] & 0x80 != 0);
280 }
281 _ => {
282 let mut refused = a;
283 refused[2] = 44;
284 return Ok((self.channel.i(&refused)?, None));
285 }
286 }
287 Ok((auto, Some(r)))
288 }
289 fn answer(&mut self, r: &Value, a: Option<&Value>) -> Result<Vec<u8>> {
290 let ca = number(r, "common_address", 65535)? as u16;
291 let mut out = vec![];
292 if r["operation"] == "command" {
293 let accepted = a
294 .is_some_and(|a| a["type"] == "iec104_command_result" && a["accepted"] == true)
295 && r["select"] != true;
296 let mut response = self.request.clone();
297 response[2] = 7 | if accepted { 0 } else { 0x40 };
298 out.extend(self.channel.i(&response)?);
299 if accepted {
300 response[2] = 10;
301 out.extend(self.channel.i(&response)?);
302 }
303 return Ok(out);
304 }
305 let points = a
306 .filter(|a| a["type"] == "iec104_measurements")
307 .and_then(|a| a["points"].as_array());
308 let good =
309 points.is_some_and(|p| p.len() <= 8 && p.iter().all(|p| point(p, ca, 20).is_ok()));
310 if r["operation"] == "interrogate" {
311 let mut confirmation = self.request.clone();
312 let accepted = good && r["qualifier"] == 20;
313 confirmation[2] = 7 | if accepted { 0 } else { 0x40 };
314 out.extend(self.channel.i(&confirmation)?);
315 if accepted {
316 for p in points.expect("checked") {
317 out.extend(self.channel.i(&point(p, ca, 20)?)?);
318 }
319 confirmation[2] = 10;
320 out.extend(self.channel.i(&confirmation)?);
321 }
322 } else if good {
323 let selected = points
324 .expect("checked")
325 .iter()
326 .find(|p| p["ioa"] == r["ioa"]);
327 if let Some(p) = selected {
328 out.extend(self.channel.i(&point(p, ca, 5)?)?);
329 } else {
330 let mut response = self.request.clone();
331 response[2] = 47;
332 out.extend(self.channel.i(&response)?);
333 }
334 } else {
335 let mut response = self.request.clone();
336 response[2] = 47;
337 out.extend(self.channel.i(&response)?);
338 }
339 Ok(out)
340 }
341}
342pub fn validate(a: &Value) -> Result<()> {
343 number(a, "common_address", 65535)?;
344 match a["type"].as_str() {
345 Some("iec104_interrogate") => {}
346 Some("iec104_read") => {
347 number(a, "ioa", 0xffffff)?;
348 }
349 Some("iec104_command") => {
350 number(a, "ioa", 0xffffff)?;
351 ensure!(
352 a["value"].is_boolean(),
353 "single command value must be boolean"
354 );
355 }
356 _ => bail!("unknown IEC 104 action"),
357 };
358 Ok(())
359}
360#[derive(Default)]
361pub struct Scanner {
362 channel: Channel,
363}
364#[async_trait::async_trait]
365impl ScannerSession for Scanner {
366 async fn idle(&mut self, s: &mut TcpStream) -> Result<Option<Value>> {
367 let f = self.channel.read(s).await?;
368 let (auto, a) = self.channel.receive(&f)?;
369 if !auto.is_empty() {
370 s.write_all(&auto).await?;
371 }
372 if let Some(a) = a {
373 s.write_all(&self.channel.s()).await?;
374 Ok(Some(json!({"success":true,"points":decode_telemetry(&a)?})))
375 } else {
376 Ok(None)
377 }
378 }
379 async fn open(&mut self, s: &mut TcpStream) -> Result<()> {
380 s.write_all(&u_frame(7)).await?;
381 loop {
382 let f = self.channel.read(s).await?;
383 let (auto, a) = self.channel.receive(&f)?;
384 ensure!(a.is_none(), "ASDU before STARTDT");
385 if !auto.is_empty() {
386 s.write_all(&auto).await?;
387 }
388 if self.channel.active {
389 return Ok(());
390 }
391 }
392 }
393 async fn exchange(&mut self, s: &mut TcpStream, a: &Value) -> Result<Value> {
394 validate(a)?;
395 let ca = number(a, "common_address", 65535)? as u16;
396 let mut points = vec![];
397 let (typ, request) = match a["type"].as_str() {
398 Some("iec104_interrogate") => (100, asdu(100, 6, 0, ca, 0, &[20])),
399 Some("iec104_read") => (
400 102,
401 asdu(102, 5, 0, ca, number(a, "ioa", 0xffffff)? as u32, &[]),
402 ),
403 _ => (
404 45,
405 asdu(
406 45,
407 6,
408 0,
409 ca,
410 number(a, "ioa", 0xffffff)? as u32,
411 &[u8::from(a["value"].as_bool().context("command value")?)],
412 ),
413 ),
414 };
415 s.write_all(&self.channel.i(&request)?).await?;
416 loop {
417 let f = self.channel.read(s).await?;
418 let (auto, response) = self.channel.receive(&f)?;
419 if !auto.is_empty() {
420 s.write_all(&auto).await?;
421 }
422 let Some(response) = response else { continue };
423 s.write_all(&self.channel.s()).await?;
424 ensure!(
425 u16::from_le_bytes([response[4], response[5]]) == ca,
426 "common address mismatch"
427 );
428 let cause = response[2] & 63;
429 if matches!(cause, 44..=47) || response[2] & 0x40 != 0 {
430 return Ok(json!({"success":false,"cause":cause}));
431 }
432 if matches!(response[0], 1 | 13) {
433 points.extend(decode_telemetry(&response)?);
434 ensure!(points.len() <= 64, "telemetry response bound");
435 if typ == 102 {
436 return Ok(json!({"success":true,"points":points}));
437 }
438 continue;
439 }
440 ensure!(
441 response[0] == typ && response[6..] == request[6..],
442 "ASDU operation correlation"
443 );
444 if cause == 10 {
445 return Ok(json!({"success":true,"points":points}));
446 }
447 ensure!(cause == 7, "unexpected confirmation cause");
448 }
449 }
450}