Grcov report - eval_subsystem

1

//! Drives the `nomisync-eval` subsystem over a real SSH connection, so the

2

//! channel shutdown is checked end to end rather than through a mock sink.

3

4

use crate::handler::{SshConnectInfo, SshdState};

5

use crate::rate_limit::{Config as LimitConfig, InstantClock, RateLimiter};

6

use russh::client::{self, Handle};

7

use russh::keys::{Algorithm, HashAlg, PrivateKey, PrivateKeyWithHashAlg, PublicKeyOrCertificate};

8

use russh::server::{RunningServerHandle, Server as _};

9

use russh::{Channel, ChannelMsg, Disconnect};

10

use server::command::ssh_key::AddSshKey;

11

use server::db::DB_POOL;

12

use sqlx::PgPool;

13

use sqlx::types::Uuid;

14

use std::net::SocketAddr;

15

use std::sync::Arc;

16

use std::time::Duration;

17

use supp_macro::local_db_sqlx_test;

18

use tokio::net::TcpListener;

19

use tokio::sync::{Mutex, oneshot};

20

use tokio::task::JoinHandle;

21

22

/// Smaller than any single response, so every response write waits on a

23

/// window adjust and the worker's sends run behind a full window.

24

const CLIENT_WINDOW: u32 = 16;

25

const FORMS: usize = 20;

26

/// Bounds a hang, not the expected run: a passing run ends at CLOSE. Kept

27

/// under tarpaulin's 60 s silence limit so a stall fails here, with a cause.

28

const DEADLINE: Duration = Duration::from_secs(45);

29

30

/// Required by `local_db_sqlx_test`, which calls it before each case.

31

1

async fn setup() {}

32

33

struct TrustingClient;

34

35

impl client::Handler for TrustingClient {

36

    type Error = russh::Error;

37

38

1

    async fn check_server_key(&mut self, _: &PublicKeyOrCertificate) -> Result<bool, Self::Error> {

39

1

        Ok(true)

40

1

41

42

43

struct Daemon {

44

    addr: SocketAddr,

45

    handle: RunningServerHandle,

46

    task: JoinHandle<std::io::Result<()>>,

47

48

49

impl Daemon {

50

1

    async fn stop(self) -> anyhow::Result<()> {

51

1

        self.handle.shutdown("test finished".to_owned());

52

1

        self.task.await??;

53

1

        Ok(())

54

1

55

56

57

#[derive(Debug, Default)]

58

struct Transcript {

59

    output: Vec<u8>,

60

    exit_status: Option<u32>,

61

    eof: bool,

62

    data_after_eof: bool,

63

    closed: bool,

64

65

66

2

fn random_key() -> anyhow::Result<PrivateKey> {

67

2

    let mut rng = rand_core::UnwrapErr(rand::rngs::SysRng);

68

2

    Ok(PrivateKey::random(&mut rng, Algorithm::Ed25519)?)

69

2

70

71

1

async fn insert_verified_user(pool: &PgPool, id: Uuid) -> anyhow::Result<()> {

72

1

    sqlx::query!(

73

        "INSERT INTO users (

74

            id, user_name, email, photo, verified, user_password,

75

            user_role, database_name, created_at

76

        ) VALUES (

77

            $1, 'eval-e2e', $2, 'default.png',

78

            TRUE, 'irrelevant', 'user', 'eval-e2e', NOW()

79

)",

80

id,

81

        format!("eval-e2e-{id}@example.test"),

82

83

1

    .execute(pool)

84

1

    .await?;

85

1

    Ok(())

86

1

87

88

1

async fn register_key(user_id: Uuid, key: &PrivateKey) -> anyhow::Result<()> {

89

1

    let public = key.public_key();

90

1

    AddSshKey::new()

91

1

        .user_id(user_id)

92

1

        .key_type(public.algorithm().as_str().to_owned())

93

1

        .key_blob(public.to_bytes()?)

94

1

        .fingerprint(public.fingerprint(HashAlg::Sha256).to_string())

95

1

        .run()

96

1

        .await?;

97

1

    Ok(())

98

1

99

100

1

async fn start_daemon() -> anyhow::Result<Daemon> {

101

1

    let listener = TcpListener::bind("127.0.0.1:0").await?;

102

1

    let addr = listener.local_addr()?;

103

1

    let mut state = SshdState {

104

1

        limiter: Arc::new(Mutex::new(RateLimiter::new(

105

1

            LimitConfig {

106

1

                max_attempts_per_ip: 6,

107

1

                max_pubkey_attempts_per_ip: 30,

108

1

                window: Duration::from_mins(1),

109

1

                account_lockout_threshold: 3,

110

1

                account_lockout_duration: Duration::from_mins(15),

111

1

},

112

1

            InstantClock,

113

1

        ))),

114

1

        connect_info: SshConnectInfo {

115

1

            hostname: addr.ip().to_string(),

116

1

            port: addr.port(),

117

1

            host_fingerprint: String::new(),

118

1

},

119

1

};

120

1

    let config = Arc::new(russh::server::Config {

121

1

        keys: vec![random_key()?],

122

1

        ..russh::server::Config::default()

123

});

124

1

    let (handle_tx, handle_rx) = oneshot::channel();

125

    // The server future borrows the listener, so it has to be built inside the

126

    // task that owns it; the shutdown handle comes back out through the channel.

127

1

    let task = tokio::spawn(async move {

128

1

        let server = state.run_on_socket(config, &listener);

129

1

        if handle_tx.send(server.handle()).is_err() {

130

            return Ok(());

131

1

132

1

        server.await

133

1

});

134

    Ok(Daemon {

135

1

        addr,

136

1

        handle: handle_rx.await?,

137

1

        task,

138

})

139

1

140

141

1

async fn open_eval_channel(

142

1

    addr: SocketAddr,

143

1

    key: PrivateKey,

144

1

) -> anyhow::Result<(Handle<TrustingClient>, Channel<client::Msg>)> {

145

1

    let config = Arc::new(client::Config {

146

1

        window_size: CLIENT_WINDOW,

147

1

        ..client::Config::default()

148

1

});

149

1

    let mut session = client::connect(config, addr, TrustingClient).await?;

150

1

    let auth = session

151

1

        .authenticate_publickey("eval-e2e", PrivateKeyWithHashAlg::new(Arc::new(key), None))

152

1

        .await?;

153

1

    anyhow::ensure!(auth.success(), "publickey auth was refused");

154

1

    let channel = session.channel_open_session().await?;

155

1

    channel.request_subsystem(true, "nomisync-eval").await?;

156

1

    Ok((session, channel))

157

1

158

159

1

async fn read_until_closed(channel: &mut Channel<client::Msg>) -> Transcript {

160

1

    let mut transcript = Transcript::default();

161

46

    while let Some(msg) = channel.wait().await {

162

46

        match msg {

163

43

            ChannelMsg::Data { data } => {

164

43

                transcript.data_after_eof |= transcript.eof;

165

43

                transcript.output.extend_from_slice(&data);

166

43

167

1

            ChannelMsg::ExitStatus { exit_status } => transcript.exit_status = Some(exit_status),

168

1

            ChannelMsg::Eof => transcript.eof = true,

169

            ChannelMsg::Close => {

170

1

                transcript.closed = true;

171

1

                break;

172

173

            _ => {}

174

175

176

1

    transcript

177

1

178

179

/// Pipes every form, half-closes the input, and reads until the daemon ends

180

/// the channel.

181

1

async fn pipe_forms_and_read(addr: SocketAddr, key: PrivateKey) -> anyhow::Result<Transcript> {

182

1

    let (session, mut channel) = open_eval_channel(addr, key).await?;

183

1

    channel.data(piped_forms().as_bytes()).await?;

184

1

    channel.eof().await?;

185

1

    let transcript = read_until_closed(&mut channel).await;

186

1

    session

187

1

        .disconnect(Disconnect::ByApplication, "done", "")

188

1

        .await?;

189

1

    Ok(transcript)

190

1

191

192

1

fn response_lines(output: &[u8]) -> Vec<String> {

193

1

    String::from_utf8_lossy(output)

194

1

        .lines()

195

1

        .map(str::to_owned)

196

1

        .collect()

197

1

198

199

1

fn answered_ids(lines: &[String]) -> Vec<usize> {

200

1

    lines

201

1

        .iter()

202

20

        .filter(|line| line.ends_with(":value 1)"))

203

20

        .filter_map(|line| line.strip_prefix("(:id ")?.split_once(' ')?.0.parse().ok())

204

1

        .collect()

205

1

206

207

1

fn piped_forms() -> String {

208

1

    (1..=FORMS)

209

20

        .map(|id| format!("(:id {id} :form (rpc-protocol-version))\n"))

210

1

        .collect()

211

1

212

213

#[local_db_sqlx_test]

214

async fn piped_forms_are_all_answered_before_the_channel_ends(pool: PgPool) -> anyhow::Result<()> {

215

    let user_id = Uuid::new_v4();

216

    insert_verified_user(&pool, user_id).await?;

217

    let key = random_key()?;

218

    register_key(user_id, &key).await?;

219

    let daemon = start_daemon().await?;

220

221

    let outcome = tokio::time::timeout(DEADLINE, pipe_forms_and_read(daemon.addr, key)).await;

222

    daemon.stop().await?;

223

    let transcript = outcome??;

224

225

    let lines = response_lines(&transcript.output);

226

    assert!(

227

        lines

228

            .iter()

229

20

            .all(|line| line.len() >= CLIENT_WINDOW as usize),

230

        "every response must fill the client window: {lines:?}"

231

);

232

    assert_eq!(answered_ids(&lines), (1..=FORMS).collect::<Vec<_>>());

233

    assert_eq!(transcript.exit_status, Some(0));

234

    assert!(transcript.eof, "{transcript:?}");

235

    assert!(!transcript.data_after_eof, "{transcript:?}");

236

    assert!(transcript.closed, "{transcript:?}");

237