pipe-audit-core 0.1.12

Universal data validation and compliance audit engine
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

use crate::connectors::{AzureConnector, Connector, GCSConnector, LocalConnector, S3Connector};
use crate::contracts::schema::Source;
use crate::logging::error::{ValidationError, ValidationResult};
use crate::profiles::Profiles;
use std::io::Read;
use url::Url;

pub async fn fetch_data_from_source(
    source: &Source,
    profiles: &Profiles,
) -> ValidationResult<Vec<u8>> {
    let location = source
        .location
        .as_ref()
        .ok_or_else(|| ValidationError::Other("Source missing location".to_string()))?;

    match source.r#type.as_str() {
        "local" => {
            let connector = LocalConnector::new();
            let mut reader = connector
                .fetch(location)
                .await
                .map_err(|e| ValidationError::Connector(e.to_string()))?;
            let mut buf = Vec::new();
            reader
                .read_to_end(&mut buf)
                .map_err(|e| ValidationError::Connector(e.to_string()))?;
            Ok(buf)
        }
        "s3" => {
            let profile_name = source
                .profile
                .as_ref()
                .ok_or_else(|| ValidationError::Other("S3 source requires profile".to_string()))?;
            let profile = profiles
                .get(profile_name)
                .ok_or_else(|| ValidationError::ProfileNotFound(profile_name.clone()))?;
            let url = Url::parse(location)
                .map_err(|_| ValidationError::Other("Invalid URL".to_string()))?;
            let connector = S3Connector::from_profile_and_url(profile, &url)
                .await
                .map_err(|e| ValidationError::Connector(e.to_string()))?;
            let mut reader = connector
                .fetch(location)
                .await
                .map_err(|e| ValidationError::Connector(e.to_string()))?;
            let mut buffer = Vec::new();
            reader
                .read_to_end(&mut buffer)
                .map_err(ValidationError::Io)?;
            Ok(buffer)
        }
        "azure" => {
            let profile_name = source.profile.as_ref().ok_or_else(|| {
                ValidationError::Other("Azure source requires profile".to_string())
            })?;
            let profile = profiles
                .get(profile_name)
                .ok_or_else(|| ValidationError::ProfileNotFound(profile_name.clone()))?;
            let url = Url::parse(location)
                .map_err(|_| ValidationError::Other("Invalid URL".to_string()))?;
            let connector = AzureConnector::from_profile_and_url(profile, &url)
                .await
                .map_err(|e| ValidationError::Connector(e.to_string()))?;
            let mut reader = connector
                .fetch(location)
                .await
                .map_err(|e| ValidationError::Connector(e.to_string()))?;
            let mut buffer = Vec::new();
            reader
                .read_to_end(&mut buffer)
                .map_err(ValidationError::Io)?;
            Ok(buffer)
        }
        "gcs" => {
            let profile_name = source
                .profile
                .as_ref()
                .ok_or_else(|| ValidationError::Other("GCS source requires profile".to_string()))?;
            let profile = profiles
                .get(profile_name)
                .ok_or_else(|| ValidationError::ProfileNotFound(profile_name.clone()))?;
            let url = Url::parse(location)
                .map_err(|_| ValidationError::Other("Invalid URL".to_string()))?;
            let connector = GCSConnector::from_profile_and_url(profile, &url)
                .await
                .map_err(|e| ValidationError::Connector(e.to_string()))?;
            let mut reader = connector
                .fetch(location)
                .await
                .map_err(|e| ValidationError::Connector(e.to_string()))?;
            let mut buffer = Vec::new();
            reader
                .read_to_end(&mut buffer)
                .map_err(ValidationError::Io)?;
            Ok(buffer)
        }
        _ => Err(ValidationError::Other(format!(
            "Unsupported source type: {}",
            source.r#type
        ))),
    }
}