filespooler 1.2.4

/*! The file header and metadata.

See the description of the file format at [`FSPrefix`].

[`crate::jobfile`] has thin wrappers around some of these functions.
*/
/*
    Copyright (C) 2022 John Goerzen <jgoerzen@complete.org>

    This program is free software: you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
    the Free Software Foundation, either version 3 of the License, or
    (at your option) any later version.

    This program is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
    GNU General Public License for more details.

    You should have received a copy of the GNU General Public License
    along with this program.  If not, see <http://www.gnu.org/licenses/>.
*/

use anyhow::ensure;
use bytes::{Buf, BufMut, Bytes, BytesMut};
use chrono::offset::LocalResult;
use chrono::prelude::*;
use crc32fast;
use rmp_serde;
use serde::{Deserialize, Serialize};
use std::ffi::OsString;
use std::io::{Read, Write};
use std::os::unix::ffi::OsStrExt;

/// The core metadata about file spooler jobs.
///
/// A file on-disk containts a [`FSPrefix`] followed by metadata.
#[derive(Serialize, Deserialize, Debug, Eq, PartialEq, Clone)]
pub struct FSMetaV1 {
    /// The sequence number of this job.
    pub seq: u64,

    /// Timestamp at creation (Unix epoch based, (seconds, nanos))
    pub ctime: (i64, u32),

    /// Command type
    pub cmd_type: CommandTypeV1,
}

/// The type of packet this is.
#[derive(Serialize, Deserialize, Debug, Eq, PartialEq, Clone)]
pub enum CommandTypeV1 {
    /// A regular command
    Command(CommandDataV1),

    /// A no-op (always succeeds)
    NOP,

    /// A failure (always fails)
    Fail,
}

/// Metadata for a command
#[derive(Serialize, Deserialize, Debug, Eq, PartialEq, Clone)]
pub struct CommandDataV1 {
    /// Parameters to pass to the command
    pub params: Vec<Vec<u8>>,

    /// Environment variables to set
    pub env: Vec<(Vec<u8>, Vec<u8>)>,
}

pub const ENV_PREFIX: &[u8] = b"FSPL_SET_";

impl FSMetaV1 {
    /// Renders in a shell-type format.  Will include the filename
    /// if available.
    pub fn render<W: Write>(
        &self,
        filename: Option<&OsString>,
        queuedir: Option<&OsString>,
        out: &mut W,
    ) -> Result<(), anyhow::Error> {
        for (key, val) in self.get_envvars(filename, queuedir).into_iter() {
            out.write_all(&key)?;
            out.write_all(b"=")?;
            out.write_all(&val)?;
            out.write_all(b"\n")?;
        }
        Ok(())
    }

    /// Get settings suitable for environment variables.  Also will include
    /// a filename if it is available.
    pub fn get_envvars(
        &self,
        filename: Option<&OsString>,
        queuedir: Option<&OsString>,
    ) -> Vec<(Vec<u8>, Vec<u8>)> {
        fn push_vu8(v: &mut Vec<(Vec<u8>, Vec<u8>)>, key: &str, value: Vec<u8>) {
            v.push((key.as_bytes().to_vec(), value));
        }

        fn push_str(v: &mut Vec<(Vec<u8>, Vec<u8>)>, key: &str, value: &str) {
            push_vu8(v, key, value.as_bytes().to_vec());
        }

        let mut retval: Vec<(Vec<u8>, Vec<u8>)> = Vec::new();

        push_str(&mut retval, "FSPL_SEQ", &self.seq.to_string());
        push_str(&mut retval, "FSPL_CTIME_SECS", &self.ctime.0.to_string());
        push_str(&mut retval, "FSPL_CTIME_NANOS", &self.ctime.1.to_string());
        push_str(
            &mut retval,
            "FSPL_CTIME_RFC3339_UTC",
            &self.get_datestring_rfc3339_utc(),
        );
        push_str(
            &mut retval,
            "FSPL_CTIME_RFC3339_LOCAL",
            &self.get_datestring_rfc3339_local(),
        );

        if let Some(filename) = filename {
            push_vu8(
                &mut retval,
                "FSPL_JOB_FILENAME",
                Vec::from(filename.as_bytes()),
            );
        }

        if let Some(queuedir) = queuedir {
            push_vu8(
                &mut retval,
                "FSPL_JOB_QUEUEDIR",
                Vec::from(queuedir.as_bytes()),
            );
        }

        if let (Some(filename), Some(queuedir)) = (filename, queuedir) {
            let mut fullpath = Vec::from(queuedir.as_bytes());
            fullpath.extend(b"/jobs/");
            fullpath.extend(filename.as_bytes());
            push_vu8(&mut retval, "FSPL_JOB_FULLPATH", fullpath);
        }

        if let CommandTypeV1::Command(cdata) = &self.cmd_type {
            for (param, num) in cdata.params.iter().zip(1..) {
                push_vu8(&mut retval, &format!("FSPL_PARAM_{}", num), param.clone());
            }

            for (param, val) in cdata.env.iter() {
                let mut key = Vec::from(ENV_PREFIX);
                key.append(&mut param.clone());
                retval.push((key, val.clone()));
            }
        }

        retval
    }

    /// Gets a chrono DateTime<Utc> for the timestamp
    pub fn get_datetime_utc(&self) -> DateTime<Utc> {
        DateTime::from_timestamp(self.ctime.0, self.ctime.1)
            .expect("Invalid timestamp in struct")
    }

    /// Gets a chrono Datetime<Local> for the timestamp
    pub fn get_datetime_local(&self) -> DateTime<Local> {
        match Local.timestamp_opt(self.ctime.0, self.ctime.1) {
            LocalResult::None => panic!("Invalid timestamp in struct"),
            LocalResult::Single(x) => x,
            LocalResult::Ambiguous(x, _) => x, // Ugh, is this even possible?
        }
    }

    /// Gets a chrono DateTime<Local> for the timestamp

    /// Renders the timestamp as RFC3339 in UTC
    pub fn get_datestring_rfc3339_utc(&self) -> String {
        let dtutc = self.get_datetime_utc();
        dtutc.to_rfc3339_opts(SecondsFormat::Secs, true)
    }

    /// Gets the local time string for the timestamp
    pub fn get_datestring_rfc3339_local(&self) -> String {
        let dtlocal = self.get_datetime_local();
        dtlocal.to_rfc3339_opts(SecondsFormat::Secs, false)
    }
}

impl Default for FSMetaV1 {
    fn default() -> Self {
        let ctime = Utc::now();
        Self {
            seq: 1,
            cmd_type: CommandTypeV1::NOP,
            ctime: (ctime.timestamp(), ctime.timestamp_subsec_nanos()),
        }
    }
}

/// The very first bytes of the file.
pub const FILETAG: &str = "*FILESPOOLER";

/**
The on-disk first few bytes of the file, consisting of:

- The [`FILETAG`]
- A u16 version
- A u64 length of the [`FSMetaV1`]
- A CRC32 of the [`FSMetaV1`]
- A CRC32 of this prefix (everything from start-of-file to immediately before this point)

After this, the FSMetaV1 can be safely read and validated based on the length and CRC.

Immediately after the FSMetaV1, the payload (if any) is present.
*/
#[derive(Debug, Eq, PartialEq, Clone)]
pub struct FSPrefix {
    pub version: u16,
    pub meta: FSMetaV1,
}

impl FSPrefix {
    pub fn prefixlen() -> usize {
        FILETAG.as_bytes().len() + 2 + 8 + 4 + 4
    }

    /// Encodes self, returning the encoded prefix and meta.
    /// You probably want [`FSPrefix::write`] instead.
    pub fn encode(&self) -> Result<(Bytes, Vec<u8>), anyhow::Error> {
        let bufsize = FSPrefix::prefixlen();
        let crcsize = bufsize - 4;

        let mut prefix = BytesMut::with_capacity(bufsize);
        prefix.put_slice(FILETAG.as_bytes());
        prefix.put_u16(self.version);

        let metavec = rmp_serde::encode::to_vec_named(&self.meta)?;
        prefix.put_u64(metavec.len() as u64);

        // Now the metacrc
        prefix.put_u32(crc32fast::hash(metavec.as_ref()));

        let prefixcrc = crc32fast::hash(&prefix[0..crcsize]);
        prefix.put_u32(prefixcrc);

        assert_eq!(bufsize, prefix.len());
        Ok((prefix.freeze(), metavec))
    }

    /// Reads, precisely, the prefix and metadata from the input reader.
    /// Does not consume any of the payload.
    pub fn read<R: Read + ?Sized>(input: &mut R) -> Result<Self, anyhow::Error> {
        let bufsize = FSPrefix::prefixlen();
        let crcsize = bufsize - 4;

        let mut prefix = BytesMut::with_capacity(bufsize);
        prefix.resize(bufsize, 0);

        input.read_exact(&mut prefix)?;
        let mut prefix = prefix.freeze();

        let crcpart = prefix.slice(0..crcsize);

        let taglen = FILETAG.as_bytes().len();

        ensure!(
            *FILETAG.as_bytes() == prefix[0..taglen],
            "Input doesn't appear to be a filespooler file"
        );

        prefix.advance(taglen);
        let version = prefix.get_u16();
        ensure!(version == 1, "Input version {} != 1", version);

        let metalen = prefix.get_u64();
        let metacrc = prefix.get_u32();
        let prefixcrc = prefix.get_u32();

        ensure!(
            prefixcrc == crc32fast::hash(&crcpart),
            "CRC-32 mismatch on file prefix"
        );

        let mut metabuf = BytesMut::with_capacity(metalen as usize);
        metabuf.resize(metalen as usize, 0u8);
        input.read_exact(&mut metabuf)?;
        assert_eq!(metabuf.len(), metalen as usize);

        ensure!(
            metacrc == crc32fast::hash(&metabuf),
            "CRC-32 mismatch on metadata"
        );

        let meta: FSMetaV1 = rmp_serde::decode::from_slice(&metabuf)?;
        Ok(Self { version: 1, meta })
    }

    /// Precisely writes the prefix and metadata to the output.  Does not
    /// emit any payload.
    pub fn write<W: Write>(&self, output: &mut W) -> Result<(), anyhow::Error> {
        let (header, metadata) = self.encode()?;
        output.write_all(&header)?;
        output.write_all(&metadata)?;
        Ok(())
    }
}

impl From<FSMetaV1> for FSPrefix {
    fn from(meta: FSMetaV1) -> Self {
        Self { version: 1, meta }
    }
}

impl From<FSPrefix> for FSMetaV1 {
    fn from(prefix: FSPrefix) -> Self {
        prefix.meta
    }
}

impl Default for FSPrefix {
    fn default() -> Self {
        Self {
            version: 1,
            meta: FSMetaV1::default(),
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_default() {
        let pref = FSPrefix::default();
        let mut buf = vec![].writer();
        pref.write(&mut buf).unwrap();

        let buf = buf.into_inner();
        let mut reader = buf.reader();
        let pref2 = FSPrefix::read(&mut reader).unwrap();

        assert_eq!(pref, pref2);
    }

    fn test_modified(offset: usize, value: u8, message: &str) {
        let pref = FSPrefix::default();
        let mut buf = vec![].writer();
        pref.write(&mut buf).unwrap();

        let mut buf = buf.into_inner();
        buf[offset] = value;

        let mut reader = buf.reader();
        let err = FSPrefix::read(&mut reader).unwrap_err();
        assert_eq!(message, format!("{}", err));
    }

    #[test]
    fn test_header() {
        test_modified(0, 2, "Input doesn't appear to be a filespooler file");
    }

    #[test]
    fn test_crc() {
        test_modified(11, 2, "Input doesn't appear to be a filespooler file");
        test_modified(13, 2, "Input version 2 != 1");
        test_modified(14, 2, "CRC-32 mismatch on file prefix");
        test_modified(28, 2, "CRC-32 mismatch on file prefix");
        test_modified(29, 2, "CRC-32 mismatch on file prefix");
        test_modified(30, 2, "CRC-32 mismatch on metadata");
        test_modified(31, 2, "CRC-32 mismatch on metadata");
    }
}