rtmp 0.6.5

rtmp library.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141

use {
    super::{
        define,
        define::SchemaVersion,
        errors::{DigestError, DigestErrorValue},
    },
    bytes::BytesMut,
    bytesio::bytes_reader::BytesReader,
    hmac::{Hmac, Mac, NewMac},
    sha2::Sha256,
};

pub struct DigestProcessor {
    reader: BytesReader,
    key: BytesMut,
}

impl DigestProcessor {
    pub fn new(data: BytesMut, key: BytesMut) -> Self {
        Self {
            reader: BytesReader::new(data),
            key,
        }
    }

    /* return validate digest and schema version*/
    pub fn read_digest(&mut self) -> Result<(BytesMut, SchemaVersion), DigestError> {
        if let Ok(digest) = self.generate_and_validate(SchemaVersion::Schema0) {
            return Ok((digest, SchemaVersion::Schema0));
        }

        let digest = self.generate_and_validate(SchemaVersion::Schema1)?;
        Ok((digest, SchemaVersion::Schema1))
    }

    pub fn generate_and_fill_digest(&mut self) -> Result<Vec<u8>, DigestError> {
        let (left_part, _, right_part) = self.cook_raw_message(SchemaVersion::Schema0)?;
        let raw_message = [left_part.clone(), right_part.clone()].concat();
        let computed_digest = self.make_digest(raw_message)?;

        let result = [left_part, computed_digest, right_part].concat();

        Ok(result)
    }

    pub fn generate_digest(&mut self) -> Result<BytesMut, DigestError> {
        let (left_part, _, right_part) = self.cook_raw_message(SchemaVersion::Schema0)?;
        let raw_message = [left_part, right_part].concat();
        let digest = self.make_digest(raw_message)?;

        Ok(digest)
    }

    fn find_digest_offset(&mut self, version: SchemaVersion) -> Result<usize, DigestError> {
        let mut digest_offset: usize = 0;

        match version {
            SchemaVersion::Schema0 => {
                digest_offset += self.reader.get(772)? as usize;
                digest_offset += self.reader.get(773)? as usize;
                digest_offset += self.reader.get(774)? as usize;
                digest_offset += self.reader.get(775)? as usize;

                digest_offset %= 728;
                digest_offset += 776;
            }
            SchemaVersion::Schema1 => {
                digest_offset += self.reader.get(8)? as usize;
                digest_offset += self.reader.get(9)? as usize;
                digest_offset += self.reader.get(10)? as usize;
                digest_offset += self.reader.get(11)? as usize;

                digest_offset %= 728;
                digest_offset += 12;
            }
            SchemaVersion::Unknown => {
                return Err(DigestError {
                    value: DigestErrorValue::UnknowSchema,
                });
            }
        }

        Ok(digest_offset)
    }
    /*
      +-----------------------------------------------------------+
      |                     764 bytes                             |
    * +--------------+-----------------------+--------------------+
    * |   left part  | digest data (32 bytes)|     right part     |
    * +--------------+-----------------------+--------------------+
    *                |
                     /
                     digest offset
        pice together the left part and right part to get the raw message.
     */
    fn cook_raw_message(
        &mut self,
        version: SchemaVersion,
    ) -> Result<(BytesMut, BytesMut, BytesMut), DigestError> {
        let digest_offset: usize = self.find_digest_offset(version)?;

        let mut new_reader = BytesReader::new(self.reader.get_remaining_bytes());

        let left_part = new_reader.read_bytes(digest_offset)?;
        let digest_data = new_reader.read_bytes(define::RTMP_DIGEST_LENGTH)?;
        let right_part = new_reader.extract_remaining_bytes();

        Ok((left_part, digest_data, right_part))
    }
    pub fn make_digest(&mut self, raw_message: Vec<u8>) -> Result<BytesMut, DigestError> {
        let mut mac = Hmac::<Sha256>::new_from_slice(&self.key[..]).unwrap();
        mac.update(&raw_message);
        let result = mac.finalize().into_bytes();

        if result.len() != define::RTMP_DIGEST_LENGTH {
            return Err(DigestError {
                value: DigestErrorValue::DigestLengthNotCorrect,
            });
        }

        let mut rv = BytesMut::new();
        rv.extend_from_slice(result.as_slice());

        Ok(rv)
    }

    fn generate_and_validate(&mut self, version: SchemaVersion) -> Result<BytesMut, DigestError> {
        let (left_part, digest_data, right_part) = self.cook_raw_message(version)?;
        let raw_message = [left_part, right_part].concat();

        let computed_digest = self.make_digest(raw_message)?;

        if digest_data == computed_digest {
            return Ok(digest_data);
        }

        Err(DigestError {
            value: DigestErrorValue::CannotGenerate,
        })
    }
}