1use crate::server::ics_support::{self as io, number, DeviceSession, ScannerSession};
3use anyhow::{bail, ensure, Context, Result};
4use serde_json::{json, Value};
5use tokio::{io::AsyncWriteExt, net::TcpStream};
6pub const MAX_FRAME: usize = 292;
7pub const MAX_APPLICATION: usize = 2048;
8pub fn crc(b: &[u8]) -> u16 {
9 let mut c = 0u16;
10 for byte in b {
11 c ^= *byte as u16;
12 for _ in 0..8 {
13 c = if c & 1 != 0 {
14 (c >> 1) ^ 0xa6bc
15 } else {
16 c >> 1
17 };
18 }
19 }
20 !c
21}
22pub fn link(control: u8, destination: u16, source: u16, payload: &[u8]) -> Vec<u8> {
23 assert!(payload.len() <= 250);
24 let mut b = vec![5, 0x64, (5 + payload.len()) as u8, control];
25 b.extend(destination.to_le_bytes());
26 b.extend(source.to_le_bytes());
27 b.extend(crc(&b).to_le_bytes());
28 for block in payload.chunks(16) {
29 b.extend(block);
30 b.extend(crc(block).to_le_bytes());
31 }
32 b
33}
34pub async fn read(s: &mut TcpStream) -> Result<Vec<u8>> {
35 io::frame(s, 10, MAX_FRAME, |h| {
36 ensure!(h[..2] == [5, 0x64] && h[2] >= 5, "invalid DNP3 link header");
37 ensure!(
38 crc(&h[..8]) == u16::from_le_bytes([h[8], h[9]]),
39 "DNP3 header CRC"
40 );
41 let n = h[2] as usize - 5;
42 Ok(10 + n + 2 * n.div_ceil(16))
43 })
44 .await
45}
46fn payload(f: &[u8]) -> Result<Vec<u8>> {
47 let n = f[2] as usize - 5;
48 let mut offset = 10;
49 let mut remaining = n;
50 let mut b = vec![];
51 while remaining > 0 {
52 let amount = remaining.min(16);
53 ensure!(offset + amount + 2 <= f.len(), "truncated DNP3 block");
54 let chunk = &f[offset..offset + amount];
55 ensure!(
56 crc(chunk) == u16::from_le_bytes([f[offset + amount], f[offset + amount + 1]]),
57 "DNP3 data CRC"
58 );
59 b.extend(chunk);
60 offset += amount + 2;
61 remaining -= amount;
62 }
63 ensure!(offset == f.len(), "trailing DNP3 data");
64 Ok(b)
65}
66#[derive(Default)]
67struct Assembly {
68 bytes: Vec<u8>,
69 next: u8,
70 active: bool,
71}
72impl Assembly {
73 fn push(&mut self, b: &[u8]) -> Result<Option<Vec<u8>>> {
74 ensure!(!b.is_empty(), "missing transport control");
75 let seq = b[0] & 63;
76 if b[0] & 0x40 != 0 {
77 ensure!(!self.active, "interleaved transport fragment");
78 self.bytes.clear();
79 self.active = true;
80 self.next = seq;
81 }
82 ensure!(
83 self.active && self.next == seq,
84 "transport sequence mismatch"
85 );
86 ensure!(
87 self.bytes.len() + b.len() - 1 <= MAX_APPLICATION,
88 "application exceeds 2048 bytes"
89 );
90 self.bytes.extend(&b[1..]);
91 self.next = (seq + 1) & 63;
92 if b[0] & 0x80 != 0 {
93 self.active = false;
94 Ok(Some(std::mem::take(&mut self.bytes)))
95 } else {
96 Ok(None)
97 }
98 }
99}
100fn frames(app: &[u8], master: bool, local: u16, remote: u16, seq: &mut u8) -> Vec<u8> {
101 let chunks = app.chunks(249);
102 let count = chunks.len();
103 let mut out = vec![];
104 for (i, chunk) in chunks.enumerate() {
105 let mut b =
106 vec![*seq | if i == 0 { 0x40 } else { 0 } | if i + 1 == count { 0x80 } else { 0 }];
107 b.extend(chunk);
108 *seq = (*seq + 1) & 63;
109 out.extend(link(if master { 0xc4 } else { 0x44 }, remote, local, &b));
110 }
111 out
112}
113fn read_headers(b: &[u8]) -> Result<Vec<Value>> {
114 let mut groups = vec![];
115 let mut offset = 0;
116 while offset < b.len() {
117 ensure!(offset + 3 <= b.len(), "short object header");
118 let (g, v, q) = (b[offset], b[offset + 1], b[offset + 2]);
119 offset += 3;
120 ensure!(
121 g == 60 && (1..=4).contains(&v) && q == 6,
122 "only all-object class 0/1/2/3 polls are supported"
123 );
124 groups.push(json!(v - 1));
125 }
126 ensure!(
127 !groups.is_empty() && groups.len() <= 4,
128 "invalid class poll"
129 );
130 Ok(groups)
131}
132fn encode_points(points: &[Value], event: bool) -> Result<Vec<u8>> {
134 ensure!(points.len() <= 64, "at most 64 points per response");
135 let mut b = vec![];
136 for point in points {
137 let kind = point["kind"].as_str().context("point kind")?;
138 let timestamp = point.get("timestamp_ms").and_then(Value::as_u64);
139 let timed = event && timestamp.is_some();
140 let (g, v) = match (kind, event, timed) {
141 ("binary", false, _) => (1, 2),
142 ("binary", true, false) => (2, 1),
143 ("binary", true, true) => (2, 2),
144 ("analog", false, _) => (30, 5),
145 ("analog", true, false) => (32, 5),
146 ("analog", true, true) => (32, 7),
147 ("counter", false, _) => (20, 1),
148 ("counter", true, false) => (22, 1),
149 ("counter", true, true) => (22, 5),
150 _ => bail!("unknown point kind"),
151 };
152 let index = number(point, "index", 65535)? as u16;
153 if event {
154 b.extend([g, v, 0x28, 1, 0]);
155 b.extend(index.to_le_bytes());
156 } else {
157 b.extend([g, v, 1]);
158 b.extend(index.to_le_bytes());
159 b.extend(index.to_le_bytes());
160 }
161 let flags = point
162 .get("flags")
163 .map(|n| n.as_u64().context("flags integer"))
164 .transpose()?
165 .unwrap_or(1);
166 ensure!(flags <= 255, "flags outside byte range");
167 match kind {
168 "binary" => {
169 let state = point["value"]
170 .as_bool()
171 .context("binary boolean required")?;
172 b.push((flags as u8 & 0x7f) | if state { 0x80 } else { 0 });
173 }
174 "analog" => {
175 let value = point["value"].as_f64().context("analog number required")?;
176 ensure!(
177 value.is_finite() && (value as f32).is_finite(),
178 "nonfinite analog"
179 );
180 b.push(flags as u8);
181 b.extend((value as f32).to_le_bytes());
182 }
183 "counter" => {
184 b.push(flags as u8);
185 b.extend((number(point, "value", u32::MAX as u64)? as u32).to_le_bytes());
186 }
187 _ => unreachable!(),
188 }
189 if timed {
190 let ms = timestamp.expect("checked");
191 ensure!(ms < 1u64 << 48, "timestamp exceeds 48 bits");
192 b.extend(&ms.to_le_bytes()[..6]);
193 }
194 }
195 ensure!(b.len() + 4 <= MAX_APPLICATION, "point response too large");
196 Ok(b)
197}
198fn decode_points(b: &[u8]) -> Result<Vec<Value>> {
199 let mut out = vec![];
200 let mut offset = 0;
201 while offset < b.len() {
202 ensure!(offset + 3 <= b.len(), "short measurement header");
203 let (g, v, q) = (b[offset], b[offset + 1], b[offset + 2]);
204 offset += 3;
205 let (count, start, prefix) = match q {
206 0x28 => {
207 ensure!(offset + 2 <= b.len(), "short count");
208 let n = u16::from_le_bytes([b[offset], b[offset + 1]]) as usize;
209 offset += 2;
210 (n, 0, true)
211 }
212 0x17 => {
213 ensure!(offset < b.len(), "short count");
214 let n = b[offset] as usize;
215 offset += 1;
216 (n, 0, false)
217 }
218 0 => {
219 ensure!(offset + 2 <= b.len(), "short range");
220 let start = b[offset] as usize;
221 let end = b[offset + 1] as usize;
222 ensure!(start <= end, "bad object range");
223 offset += 2;
224 (end - start + 1, start, false)
225 }
226 1 => {
227 ensure!(offset + 4 <= b.len(), "short range");
228 let start = u16::from_le_bytes([b[offset], b[offset + 1]]) as usize;
229 let end = u16::from_le_bytes([b[offset + 2], b[offset + 3]]) as usize;
230 ensure!(start <= end, "bad object range");
231 offset += 4;
232 (end - start + 1, start, false)
233 }
234 _ => bail!("unsupported object qualifier"),
235 };
236 ensure!(
237 count <= 128 && out.len() + count <= 128,
238 "measurement count exceeds bound"
239 );
240 let (kind, width, timed) = match (g, v) {
241 (1, 2) | (2, 1) => ("binary", 1, false),
242 (2, 2) => ("binary", 1, true),
243 (30, 5) | (32, 5) => ("analog", 5, false),
244 (32, 7) => ("analog", 5, true),
245 (20, 1) | (22, 1) => ("counter", 5, false),
246 (22, 5) => ("counter", 5, true),
247 (30, 1) | (32, 1) => ("analog_int", 5, false),
248 _ => bail!("unsupported measurement group/variation {g}/{v}"),
249 };
250 for index in 0..count {
251 let idx = if prefix {
252 ensure!(offset + 2 <= b.len(), "missing point index");
253 let i = u16::from_le_bytes([b[offset], b[offset + 1]]) as usize;
254 offset += 2;
255 i
256 } else if q == 0x17 {
257 ensure!(offset < b.len(), "missing point index");
258 let i = b[offset] as usize;
259 offset += 1;
260 i
261 } else {
262 start + index
263 };
264 ensure!(
265 offset + width + if timed { 6 } else { 0 } <= b.len(),
266 "truncated point"
267 );
268 let flags = b[offset];
269 let value = match kind {
270 "binary" => json!(flags & 0x80 != 0),
271 "analog" => {
272 let n = f32::from_le_bytes(b[offset + 1..offset + 5].try_into()?);
273 ensure!(n.is_finite(), "nonfinite wire value");
274 json!(n)
275 }
276 "analog_int" => json!(i32::from_le_bytes(b[offset + 1..offset + 5].try_into()?)),
277 _ => json!(u32::from_le_bytes(b[offset + 1..offset + 5].try_into()?)),
278 };
279 offset += width;
280 let mut point = json!({"kind":if kind=="analog_int"{"analog"}else{kind},"index":idx,"flags":flags,"value":value,"event":matches!(g,2|22|32)});
281 if timed {
282 let mut time = [0; 8];
283 time[..6].copy_from_slice(&b[offset..offset + 6]);
284 offset += 6;
285 point["timestamp_ms"] = json!(u64::from_le_bytes(time));
286 }
287 out.push(point);
288 }
289 }
290 Ok(out)
291}
292pub struct Device {
293 assembly: Assembly,
294 transport: u8,
295 local: u16,
296 remote: u16,
297 last: Option<(Vec<u8>, Vec<u8>)>,
298 confirm: Option<u8>,
299 request: Vec<u8>,
300}
301impl Default for Device {
302 fn default() -> Self {
303 Self {
304 assembly: Assembly::default(),
305 transport: 0,
306 local: 10,
307 remote: 1,
308 last: None,
309 confirm: None,
310 request: vec![],
311 }
312 }
313}
314#[async_trait::async_trait]
315impl DeviceSession for Device {
316 async fn read(&mut self, s: &mut TcpStream) -> Result<Vec<u8>> {
317 read(s).await
318 }
319 fn receive(&mut self, f: &[u8]) -> Result<(Vec<u8>, Option<Value>)> {
320 ensure!(
321 f.len() >= 10
322 && u16::from_le_bytes([f[4], f[5]]) == self.local
323 && u16::from_le_bytes([f[6], f[7]]) == self.remote
324 && f[3] & 0xc0 == 0xc0,
325 "DNP3 link addressing/direction"
326 );
327 let b = payload(f)?;
328 match f[3] & 15 {
329 0 => {
330 ensure!(b.is_empty(), "reset link payload");
331 self.assembly = Assembly::default();
332 return Ok((link(0, self.remote, self.local, &[]), None));
333 }
334 9 => {
335 ensure!(b.is_empty(), "link status payload");
336 return Ok((link(0x0b, self.remote, self.local, &[]), None));
337 }
338 4 => {}
339 _ => bail!("unsupported confirmed link service"),
340 }
341 let Some(app) = self.assembly.push(&b)? else {
342 return Ok((vec![], None));
343 };
344 ensure!(
345 app.len() >= 2 && app[0] & 0xc0 == 0xc0 && app[0] & 0x10 == 0,
346 "fragmented/unsolicited app requests unsupported"
347 );
348 let seq = app[0] & 15;
349 if app[1] == 0 {
350 ensure!(
351 app.len() == 2 && self.confirm == Some(seq),
352 "unexpected application confirmation"
353 );
354 self.confirm = None;
355 return Ok((vec![], None));
356 }
357 if let Some((last, reply)) = &self.last {
358 if last == &app {
359 return Ok((reply.clone(), None));
360 }
361 }
362 if matches!(app[1], 20 | 21) {
363 let reply = frames(
364 &[0xc0 | seq, 0x81, 0, if app[1] == 20 { 1 } else { 0 }],
365 false,
366 self.local,
367 self.remote,
368 &mut self.transport,
369 );
370 self.last = Some((app, reply.clone()));
371 return Ok((reply, None));
372 }
373 self.request = app.clone();
374 let mut request = json!({"sequence":seq,"function":app[1]});
375 match app[1] {
376 1 => match read_headers(&app[2..]) {
377 Ok(classes) => {
378 request["operation"] = json!("poll");
379 request["classes"] = json!(classes)
380 }
381 Err(_) => {
382 let reply = frames(
383 &[0xc0 | seq, 0x81, 0, 2],
384 false,
385 self.local,
386 self.remote,
387 &mut self.transport,
388 );
389 return Ok((reply, None));
390 }
391 },
392 5 => {
393 let p = &app[2..];
394 ensure!(
395 p.len() >= 4 && p[..2] == [12, 1],
396 "only CROB direct-operate supported"
397 );
398 let (base, index) = match p[2] {
399 0x17 => {
400 ensure!(p.len() == 16 && p[3] == 1, "one 8-bit CROB index required");
401 (5, p[4] as u16)
402 }
403 0x28 => {
404 ensure!(
405 p.len() == 18 && p[3..5] == [1, 0],
406 "one 16-bit CROB index required"
407 );
408 (7, u16::from_le_bytes([p[5], p[6]]))
409 }
410 _ => bail!("unsupported CROB qualifier"),
411 };
412 ensure!(
413 (1..=4).contains(&p[base]) && p[base + 1] > 0 && p[base + 10] == 0,
414 "invalid CROB fields"
415 );
416 request["operation"] = json!("control");
417 request["index"] = json!(index);
418 request["code"] = json!(p[base]);
419 request["count"] = json!(p[base + 1]);
420 request["on_ms"] = json!(u32::from_le_bytes(p[base + 2..base + 6].try_into()?));
421 request["off_ms"] = json!(u32::from_le_bytes(p[base + 6..base + 10].try_into()?));
422 }
423 _ => {
424 let reply = frames(
425 &[0xc0 | seq, 0x81, 0, 1],
426 false,
427 self.local,
428 self.remote,
429 &mut self.transport,
430 );
431 return Ok((reply, None));
432 }
433 }
434 Ok((vec![], Some(request)))
435 }
436 fn answer(&mut self, r: &Value, a: Option<&Value>) -> Result<Vec<u8>> {
437 let seq = number(r, "sequence", 15)? as u8;
438 let mut app = vec![0xc0 | seq, 0x81, 0, 0];
439 let mut events = false;
440 if r["operation"] == "poll" {
441 if let Some(a) = a.filter(|a| a["type"] == "dnp3_measurements") {
442 let points = a["points"].as_array().context("points array required")?;
443 let selected = r["classes"].as_array().context("classes")?;
444 let mut static_points = vec![];
445 let mut event_points = vec![];
446 for point in points {
447 if point.get("class").and_then(Value::as_u64).unwrap_or(0) == 0 {
448 if selected.contains(&json!(0)) {
449 static_points.push(point.clone());
450 }
451 } else if selected.contains(&point["class"]) {
452 event_points.push(point.clone());
453 }
454 }
455 let built = (|| -> Result<Vec<u8>> {
456 let mut b = encode_points(&static_points, false)?;
457 b.extend(encode_points(&event_points, true)?);
458 ensure!(b.len() + 4 <= MAX_APPLICATION, "response budget");
459 Ok(b)
460 })();
461 match built {
462 Ok(b) => {
463 events = !event_points.is_empty();
464 app.extend(b);
465 }
466 Err(_) => app[3] = 4,
467 }
468 } else {
469 app[3] = 4;
470 }
471 } else {
472 let original = &self.request;
473 app.extend(&original[2..]);
474 let status = a
475 .filter(|a| a["type"] == "dnp3_control_result")
476 .and_then(|a| a["status"].as_str());
477 let last = app.len() - 1;
478 app[last] = match status {
479 Some("success") => 0,
480 Some("not_supported") => 4,
481 Some("out_of_range") => 12,
482 _ => 6,
483 };
484 }
485 if events {
486 app[0] |= 0x20;
487 self.confirm = Some(seq);
488 }
489 let reply = frames(&app, false, self.local, self.remote, &mut self.transport);
490 let request = self.request.clone();
491 self.last = Some((request, reply.clone()));
492 Ok(reply)
493 }
494}
495pub fn validate(a: &Value) -> Result<()> {
496 match a["type"].as_str() {
497 Some("dnp3_poll") => {
498 let classes = a["classes"].as_array().context("classes required")?;
499 ensure!(
500 !classes.is_empty()
501 && classes.len() <= 4
502 && classes.iter().all(|v| v.as_u64().is_some_and(|v| v <= 3)),
503 "classes must be 0..3"
504 );
505 }
506 Some("dnp3_control") => {
507 number(a, "index", 65535)?;
508 let code = number(a, "code", 255)?;
509 ensure!((1..=4).contains(&code), "unsupported CROB code");
510 let count = number(a, "count", 255)?;
511 ensure!(count > 0, "positive control count required");
512 number(a, "on_ms", u32::MAX as u64)?;
513 number(a, "off_ms", u32::MAX as u64)?;
514 }
515 _ => bail!("unknown DNP3 action"),
516 };
517 Ok(())
518}
519#[derive(Default)]
520pub struct Scanner {
521 assembly: Assembly,
522 transport: u8,
523 sequence: u8,
524}
525#[async_trait::async_trait]
526impl ScannerSession for Scanner {
527 async fn open(&mut self, s: &mut TcpStream) -> Result<()> {
528 s.write_all(&link(0xc9, 10, 1, &[])).await?;
529 let f = read(s).await?;
530 ensure!(
531 f[3] == 0x0b && f[4..8] == [1, 0, 10, 0] && payload(&f)?.is_empty(),
532 "DNP3 link status refused"
533 );
534 Ok(())
535 }
536 async fn exchange(&mut self, s: &mut TcpStream, a: &Value) -> Result<Value> {
537 validate(a)?;
538 self.sequence = (self.sequence + 1) & 15;
539 let control = a["type"] == "dnp3_control";
540 let mut app = vec![0xc0 | self.sequence, if control { 5 } else { 1 }];
541 if control {
542 app.extend([12, 1, 0x28, 1, 0]);
543 app.extend((number(a, "index", 65535)? as u16).to_le_bytes());
544 app.extend([
545 number(a, "code", 255)? as u8,
546 number(a, "count", 255)? as u8,
547 ]);
548 app.extend((number(a, "on_ms", u32::MAX as u64)? as u32).to_le_bytes());
549 app.extend((number(a, "off_ms", u32::MAX as u64)? as u32).to_le_bytes());
550 app.push(0);
551 } else {
552 for class in a["classes"].as_array().context("classes")? {
553 app.extend([60, class.as_u64().expect("validated") as u8 + 1, 6]);
554 }
555 }
556 s.write_all(&frames(&app, true, 1, 10, &mut self.transport))
557 .await?;
558 loop {
559 let f = read(s).await?;
560 ensure!(
561 f[3] == 0x44 && f[4..8] == [1, 0, 10, 0],
562 "response link direction/addresses"
563 );
564 let Some(response) = self.assembly.push(&payload(&f)?)? else {
565 continue;
566 };
567 ensure!(
568 response.len() >= 4
569 && response[1] == 0x81
570 && response[0] & 15 == self.sequence
571 && response[0] & 0xc0 == 0xc0,
572 "application response correlation"
573 );
574 if response[0] & 0x20 != 0 {
575 s.write_all(&frames(
576 &[0xc0 | self.sequence, 0],
577 true,
578 1,
579 10,
580 &mut self.transport,
581 ))
582 .await?;
583 }
584 let iin = u16::from_le_bytes([response[2], response[3]]);
585 if iin & 0x0700 != 0 {
586 return Ok(json!({"success":false,"iin":iin}));
587 }
588 if control {
589 ensure!(
590 response.len() == app.len() + 2
591 && response[4..response.len() - 1] == app[2..app.len() - 1],
592 "control echo mismatch"
593 );
594 let status = *response.last().context("status")?;
595 return Ok(json!({"success":status==0,"status":status}));
596 }
597 return Ok(json!({"success":true,"iin":iin,"points":decode_points(&response[4..])?}));
598 }
599 }
600}