netget/server/nntp/
extensions.rs

1//! NNTP posting, AUTHINFO USER/PASS and article-feed transactions.
2use crate::llm::actions::{client_trait::ClientActionResult, ActionDefinition};
3use crate::server::p2p_support::{action, parameter};
4use anyhow::{ensure, Context, Result};
5use serde_json::{json, Value};
6use std::sync::Arc;
7use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt, BufReader};
8pub const MAX_ARTICLE: usize = 8 * 1024 * 1024;
9pub fn field(text: &str) -> Result<&str> {
10    ensure!(
11        !text.is_empty() && text.len() <= 4096 && !text.contains(['\r', '\n', '\0']),
12        "invalid NNTP field"
13    );
14    Ok(text)
15}
16pub fn message_id(text: &str) -> Result<&str> {
17    field(text)?;
18    ensure!(
19        text.starts_with('<')
20            && text.ends_with('>')
21            && !text.bytes().any(|b| b <= 32)
22            && text.contains('@'),
23        "invalid message ID"
24    );
25    Ok(text)
26}
27pub fn dot_block(text: &str) -> Result<Vec<u8>> {
28    ensure!(text.len() <= MAX_ARTICLE, "article too large");
29    let normalized = text.replace("\r\n", "\n");
30    ensure!(
31        !normalized.contains(['\r', '\0']),
32        "invalid article line ending"
33    );
34    let mut out = Vec::new();
35    for line in normalized.split_terminator('\n') {
36        ensure!(line.len() <= 64 * 1024 - 2, "article line too long");
37        if line.starts_with('.') {
38            out.push(b'.');
39        }
40        out.extend_from_slice(line.as_bytes());
41        out.extend_from_slice(b"\r\n");
42    }
43    ensure!(out.len() <= MAX_ARTICLE, "encoded article too large");
44    out.extend_from_slice(b".\r\n");
45    Ok(out)
46}
47pub fn article(v: &Value) -> Result<Vec<u8>> {
48    let headers = v["headers"]
49        .as_object()
50        .context("headers object required")?;
51    let mut text = String::new();
52    ensure!(!headers.is_empty() && headers.len() <= 128, "header count");
53    for (key, val) in headers {
54        ensure!(
55            !key.is_empty() && key.bytes().all(|b| b.is_ascii_alphanumeric() || b == b'-'),
56            "invalid header name"
57        );
58        text.push_str(key);
59        text.push_str(": ");
60        text.push_str(field(val.as_str().context("header value")?)?);
61        text.push('\n');
62    }
63    text.push('\n');
64    text.push_str(v["body"].as_str().context("body required")?);
65    dot_block(&text)
66}
67pub fn client_actions() -> Vec<ActionDefinition> {
68    let mut list = vec![
69        action(
70            "nntp_capabilities",
71            "Discover server capabilities",
72            vec![],
73            json!({"type":"nntp_capabilities"}),
74        ),
75        action(
76            "nntp_mode_stream",
77            "Enter streaming-feed mode",
78            vec![],
79            json!({"type":"nntp_mode_stream"}),
80        ),
81        action(
82            "nntp_check",
83            "Offer an article message ID",
84            vec![parameter(
85                "message_id",
86                "string",
87                "Article message ID in angle brackets, such as <article@example.org>",
88                true,
89            )],
90            json!({"type":"nntp_check","message_id":"<test@localhost>"}),
91        ),
92        action(
93            "nntp_authenticate",
94            "Authenticate over verified TLS",
95            vec![
96                parameter("username", "string", "Account username", true),
97                parameter("password", "string", "Account password", true),
98            ],
99            json!({"type":"nntp_authenticate","username":"user","password":"secret"}),
100        ),
101    ];
102    for name in ["nntp_ihave", "nntp_takethis"] {
103        list.push(action(name,"Feed an article to another news server",vec![parameter("message_id","string","RFC message ID matching the article",true),parameter("headers","object","Article header fields",true),parameter("body","string","Article body text; line normalization and dot stuffing are automatic",true)],json!({"type":name,"message_id":"<test@localhost>","headers":{"Message-ID":"<test@localhost>","From":"test@localhost","Newsgroups":"misc.test","Subject":"Test"},"body":"Hello"})));
104    }
105    list
106}
107pub fn client_action(v: &Value) -> Result<Option<ClientActionResult>> {
108    let name = v["type"].as_str().context("type")?;
109    let command = match name {
110        "nntp_capabilities" => Some("CAPABILITIES".to_string()),
111        "nntp_mode_stream" => Some("MODE STREAM".to_string()),
112        "nntp_check" => Some(format!(
113            "CHECK {}",
114            message_id(v["message_id"].as_str().context("message_id")?)?
115        )),
116        "nntp_authenticate" => {
117            field(v["username"].as_str().context("username")?)?;
118            field(v["password"].as_str().context("password")?)?;
119            return Ok(Some(ClientActionResult::Custom {
120                name: name.into(),
121                data: v.clone(),
122            }));
123        }
124        "nntp_ihave" | "nntp_takethis" => {
125            let id = message_id(v["message_id"].as_str().context("message_id")?)?;
126            article(v)?;
127            ensure!(
128                v["headers"]
129                    .as_object()
130                    .context("headers")?
131                    .iter()
132                    .any(|(k, val)| k.eq_ignore_ascii_case("message-id") && val == id),
133                "article message ID mismatch"
134            );
135            return Ok(Some(ClientActionResult::Custom {
136                name: name.into(),
137                data: v.clone(),
138            }));
139        }
140        _ => return Ok(None),
141    };
142    Ok(command.map(|command| ClientActionResult::Custom {
143        name: "nntp_command".into(),
144        data: json!({"command":command}),
145    }))
146}
147pub fn server_actions() -> Vec<ActionDefinition> {
148    [
149        "nntp_auth_result",
150        "nntp_check_result",
151        "nntp_article_result",
152    ]
153    .into_iter()
154    .map(|name| {
155        action(
156            name,
157            "Accept or refuse this authentication, offer or received article",
158            vec![parameter(
159                "accepted",
160                "boolean",
161                "Accept this request; omitted or false refuses",
162                true,
163            )],
164            json!({"type":name,"accepted":false}),
165        )
166    })
167    .collect()
168}
169#[derive(Default)]
170pub struct State {
171    pub authenticated: bool,
172    pub secure: bool,
173    pub username: Option<String>,
174    pub streaming: bool,
175    pub require_auth: bool,
176    pub posting_allowed: bool,
177}
178pub struct ContextRefs<'a> {
179    pub llm: &'a crate::llm::OllamaClient,
180    pub state: &'a Arc<crate::state::AppState>,
181    pub server: crate::state::ServerId,
182    pub peer: crate::server::connection::ConnectionId,
183    pub protocol: &'a super::NntpProtocol,
184}
185async fn decision(ctx: &ContextRefs<'_>, event: Value, kind: &str) -> bool {
186    let event = crate::protocol::Event::new(&super::actions::NNTP_COMMAND_RECEIVED_EVENT, event);
187    let Ok(result) = crate::llm::action_helper::call_llm(
188        ctx.llm,
189        ctx.state,
190        ctx.server,
191        Some(ctx.peer),
192        &event,
193        ctx.protocol,
194    )
195    .await
196    else {
197        return false;
198    };
199    result
200        .protocol_results
201        .iter()
202        .find_map(|v| {
203            if let crate::llm::ActionResult::Custom { name, data } = v {
204                (name == kind).then(|| data["accepted"].as_bool().unwrap_or(false))
205            } else {
206                None
207            }
208        })
209        .unwrap_or(false)
210}
211async fn write<W: AsyncWrite + Unpin>(
212    writer: &Arc<tokio::sync::Mutex<W>>,
213    text: &str,
214) -> Result<()> {
215    tokio::time::timeout(std::time::Duration::from_secs(10), async {
216        writer.lock().await.write_all(text.as_bytes()).await
217    })
218    .await??;
219    Ok(())
220}
221/// Return true when an extension consumed the command (and its article, when present).
222pub async fn handle<R: AsyncRead + Unpin, W: AsyncWrite + Unpin>(
223    reader: &mut BufReader<R>,
224    writer: &Arc<tokio::sync::Mutex<W>>,
225    line: &str,
226    state: &mut State,
227    ctx: &ContextRefs<'_>,
228) -> Result<bool> {
229    let words: Vec<_> = line.split_whitespace().collect();
230    let verb = words
231        .first()
232        .map(|v| v.to_ascii_uppercase())
233        .unwrap_or_default();
234    if verb == "AUTHINFO" {
235        let mut auth = line.splitn(3, ' ');
236        auth.next();
237        let mode = auth.next().unwrap_or("").to_ascii_uppercase();
238        let value = auth.next().unwrap_or("");
239        match mode.as_str() {
240            "USER" if !value.is_empty() => {
241                if !state.secure {
242                    write(writer, "483 TLS required\r\n").await?;
243                } else {
244                    state.username = Some(field(value)?.into());
245                    write(writer, "381 Password required\r\n").await?;
246                }
247            }
248            "PASS" if !value.is_empty() => {
249                if !state.secure {
250                    write(writer, "483 TLS required\r\n").await?;
251                } else if let Some(user) = state.username.take() {
252                    let accepted=decision(ctx,json!({"command":"AUTHINFO PASS","operation":"authenticate","username":user,"password":field(value)?,"answer_with":"nntp_auth_result"}),"nntp_auth_result").await;
253                    state.authenticated = accepted;
254                    write(
255                        writer,
256                        if accepted {
257                            "281 Authentication accepted\r\n"
258                        } else {
259                            "481 Authentication rejected\r\n"
260                        },
261                    )
262                    .await?;
263                } else {
264                    write(writer, "482 Send AUTHINFO USER first\r\n").await?;
265                }
266            }
267            _ => write(writer, "501 Invalid AUTHINFO syntax\r\n").await?,
268        }
269        return Ok(true);
270    }
271    if state.require_auth
272        && !state.authenticated
273        && !matches!(verb.as_str(), "CAPABILITIES" | "QUIT")
274    {
275        write(writer, "480 Authentication required\r\n").await?;
276        return Ok(true);
277    }
278    if verb == "MODE" && words.len() == 2 && words[1].eq_ignore_ascii_case("STREAM") {
279        state.streaming = true;
280        write(writer, "203 Streaming permitted\r\n").await?;
281        return Ok(true);
282    }
283    if verb == "CHECK" {
284        if !state.streaming {
285            write(writer, "500 Enter MODE STREAM first\r\n").await?;
286            return Ok(true);
287        }
288        if words.len() != 2 || message_id(words[1]).is_err() {
289            write(writer, "501 Invalid message ID\r\n").await?;
290            return Ok(true);
291        }
292        let accepted=decision(ctx,json!({"command":line,"operation":"check","message_id":words[1],"answer_with":"nntp_check_result"}),"nntp_check_result").await;
293        write(
294            writer,
295            &format!("{} {}\r\n", if accepted { 238 } else { 438 }, words[1]),
296        )
297        .await?;
298        return Ok(true);
299    }
300    if verb == "MODE"
301        && words
302            .get(1)
303            .is_some_and(|v| v.eq_ignore_ascii_case("READER"))
304    {
305        state.streaming = false;
306    }
307    if !matches!(verb.as_str(), "POST" | "IHAVE" | "TAKETHIS") {
308        return Ok(false);
309    }
310    if verb == "POST" && !state.posting_allowed {
311        write(writer, "440 Posting prohibited\r\n").await?;
312        return Ok(true);
313    }
314    if (verb == "POST" && words.len() != 1)
315        || (verb != "POST" && (words.len() != 2 || message_id(words[1]).is_err()))
316    {
317        write(writer, "501 Invalid article command\r\n").await?;
318        return Ok(true);
319    }
320    let id = if verb == "POST" {
321        None
322    } else {
323        Some(message_id(words[1])?)
324    };
325    if verb == "TAKETHIS" {
326        if !state.streaming {
327            write(writer, "500 Enter MODE STREAM first\r\n").await?;
328            return Ok(true);
329        }
330    } else {
331        write(
332            writer,
333            if verb == "POST" {
334                "340 Send article\r\n"
335            } else {
336                "335 Send article\r\n"
337            },
338        )
339        .await?;
340    }
341    let data = crate::client::response_reader::read_dot_response(reader, String::new()).await?;
342    let article = data.strip_prefix('\n').unwrap_or(&data);
343    let valid_id = id.is_none_or(|id| {
344        article.lines().take_while(|l| !l.is_empty()).any(|l| {
345            l.split_once(':').is_some_and(|(key, value)| {
346                key.eq_ignore_ascii_case("message-id") && value.trim() == id
347            })
348        })
349    });
350    let accepted=valid_id&&decision(ctx,json!({"command":line,"operation":verb.to_ascii_lowercase(),"article":article,"message_id":id,"answer_with":"nntp_article_result"}),"nntp_article_result").await;
351    let code = match (verb.as_str(), accepted) {
352        ("POST", true) => 240,
353        ("POST", false) => 441,
354        ("IHAVE", true) => 235,
355        ("IHAVE", false) => 437,
356        ("TAKETHIS", true) => 239,
357        _ => 439,
358    };
359    write(
360        writer,
361        &format!(
362            "{code} {}\r\n",
363            id.unwrap_or(if accepted {
364                "Article accepted"
365            } else {
366                "Article rejected"
367            })
368        ),
369    )
370    .await?;
371    Ok(true)
372}
netget/server/nntp/extensions.rs

netget/server/nntp/
extensions.rs
