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

/* -------------------------------------------------------------------------- *\
 *        Apache 2.0 License Copyright © 2022-2023 The Aurae Authors          *
 *                                                                            *
 *                +--------------------------------------------+              *
 *                |   █████╗ ██╗   ██╗██████╗  █████╗ ███████╗ |              *
 *                |  ██╔══██╗██║   ██║██╔══██╗██╔══██╗██╔════╝ |              *
 *                |  ███████║██║   ██║██████╔╝███████║█████╗   |              *
 *                |  ██╔══██║██║   ██║██╔══██╗██╔══██║██╔══╝   |              *
 *                |  ██║  ██║╚██████╔╝██║  ██║██║  ██║███████╗ |              *
 *                |  ╚═╝  ╚═╝ ╚═════╝ ╚═╝  ╚═╝╚═╝  ╚═╝╚══════╝ |              *
 *                +--------------------------------------------+              *
 *                                                                            *
 *                         Distributed Systems Runtime                        *
 *                                                                            *
 * -------------------------------------------------------------------------- *
 *                                                                            *
 *   Licensed under the Apache License, Version 2.0 (the "License");          *
 *   you may not use this file except in compliance with the License.         *
 *   You may obtain a copy of the License at                                  *
 *                                                                            *
 *       http://www.apache.org/licenses/LICENSE-2.0                           *
 *                                                                            *
 *   Unless required by applicable law or agreed to in writing, software      *
 *   distributed under the License is distributed on an "AS IS" BASIS,        *
 *   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. *
 *   See the License for the specific language governing permissions and      *
 *   limitations under the License.                                           *
 *                                                                            *
\* -------------------------------------------------------------------------- */
use anyhow::{anyhow, Context};
pub(crate) use cell_system_runtime::CellSystemRuntime;
pub(crate) use container_system_runtime::ContainerSystemRuntime;
pub(crate) use daemon_system_runtime::DaemonSystemRuntime;
pub(crate) use pid1_system_runtime::Pid1SystemRuntime;
use std::{
    net::SocketAddr,
    os::unix::prelude::PermissionsExt,
    path::{Path, PathBuf},
};
use tokio::net::{TcpListener, UnixListener};
use tokio_stream::wrappers::{TcpListenerStream, UnixListenerStream};
use tonic::async_trait;
use tracing::{info, trace};

use super::{fs::FsError, logging::LoggingError, network::NetworkError};

mod cell_system_runtime;
mod container_system_runtime;
mod daemon_system_runtime;
mod pid1_system_runtime;

#[derive(thiserror::Error, Debug)]
pub(crate) enum SystemRuntimeError {
    #[error(transparent)]
    FsError(#[from] FsError),
    #[error(transparent)]
    Logging(#[from] LoggingError),
    #[error(transparent)]
    Network(#[from] NetworkError),
    #[error(transparent)]
    AddrParse(#[from] std::net::AddrParseError),
    #[error(transparent)]
    Io(#[from] std::io::Error),
    #[error(transparent)]
    Other(#[from] anyhow::Error),
}

/// A [SocketStream] can represent either a TCP or Unix socket stream.
#[derive(Debug)]
pub enum SocketStream {
    /// Contains a stream for listening over a TCP socket.
    Tcp(TcpListenerStream),

    /// Contains a stream for listening over a Unix socket.
    Unix(UnixListenerStream),
}

#[async_trait]
pub(crate) trait SystemRuntime {
    async fn init(
        self,
        verbose: bool,
        socket_address: Option<String>,
    ) -> Result<SocketStream, SystemRuntimeError>;
}

async fn create_unix_socket_stream(
    socket_path: PathBuf,
) -> Result<SocketStream, SystemRuntimeError> {
    let _ = std::fs::remove_file(&socket_path);
    let sock_path = Path::new(&socket_path).parent().ok_or_else(|| {
        anyhow!("not a valid socket path: {:?}", &socket_path)
    })?;
    // Create socket directory
    tokio::fs::create_dir_all(sock_path).await.with_context(|| {
        format!(
            "Failed to create directory for socket: {}",
            socket_path.display()
        )
    })?;
    trace!("User Access Socket dir created: {}", sock_path.display());

    let sock = UnixListener::bind(&socket_path)?;

    // We set the mode to 766 for the Unix domain socket.
    // This is what allows non-root users to dial the socket
    // and authenticate with mTLS.
    trace!("Setting socket mode {} -> 766", &socket_path.display());
    std::fs::set_permissions(
        &socket_path,
        std::fs::Permissions::from_mode(0o766),
    )?;
    info!("User Access Socket Created: {}", socket_path.display());

    Ok(SocketStream::Unix(UnixListenerStream::new(sock)))
}

async fn create_tcp_socket_stream(
    socket_addr: SocketAddr,
) -> Result<SocketStream, SystemRuntimeError> {
    trace!("creating tcp stream for {:?}", socket_addr);
    let sock = TcpListener::bind(&socket_addr).await?;
    info!("TCP Access Socket created: {:?}", socket_addr);
    Ok(SocketStream::Tcp(TcpListenerStream::new(sock)))
}