michal/tit
Browse tree · Show commit · Download archive
Blob: src/ssh.rs
use std::borrow::Cow;
use std::collections::{HashMap, HashSet};
use std::net::{IpAddr, Ipv4Addr, SocketAddr};
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::sync::{Arc, RwLock};
use std::time::{Duration, Instant};
use rand::rng;
use russh::server::{Auth, ChannelOpenHandle, Handler, Msg, Server, Session};
use russh::{Channel, ChannelId, MethodKind, MethodSet, Preferred, Pty};
use ssh_key::{Algorithm, EcdsaCurve, PrivateKey, PublicKey};
use thiserror::Error;
use tokio::io::AsyncWriteExt;
use tokio::net::TcpListener;
use tokio::sync::{OwnedSemaphorePermit, Semaphore, oneshot};
use tokio::task::JoinHandle;
use crate::auth::SshPublicKey;
use crate::git::packetline::{MAX_REQUEST_BYTES, Packet, decode, encode_data, first_flush_end};
use crate::git::receive_pack::{ReceivePack, ReceivePackError};
use crate::git::transport::{GitRepositories, GitSshService};
use crate::git::upload_pack::{ProtocolVersion, UploadPack, UploadPackError};
use crate::issue::{IssueError, IssueService, MAX_BODY_BYTES, MAX_TITLE_BYTES};
use crate::policy::RepositoryOperation;
use crate::pull_request::{PullRequestError, PullRequestService};
use crate::rate_limit::AttemptLimiter;
use crate::repository::{RepositoryService, RepositoryServiceError};
use crate::store::{Store, StoreError};
use crate::telemetry::Telemetry;
const VERSION_COMMAND: &[u8] = b"tit --version";
const HELP_COMMAND: &[u8] = b"help";
const AUTH_COMMAND_PREFIX: &[u8] = b"auth ";
const GIT_PROTOCOL_VARIABLE: &str = "GIT_PROTOCOL";
const MAX_RECEIVE_PACK_BYTES: u64 = 128 * 1024 * 1024;
const MAX_REPOSITORY_COMMAND_BYTES: usize = 512;
const MAX_ISSUE_COMMAND_BYTES: usize = 512;
const MAX_PULL_REQUEST_COMMAND_BYTES: usize = 512;
const MAX_ISSUE_INPUT_BYTES: usize = MAX_TITLE_BYTES + 1 + MAX_BODY_BYTES;
const SSH_ATTEMPTS_PER_MINUTE: usize = 30;
const MAX_SSH_CLIENTS: usize = 4096;
const MAX_GIT_CHANNELS_PER_CONNECTION: usize = 4;
const MAX_GIT_CHANNELS_GLOBAL: usize = 64;
const GIT_CHANNEL_INACTIVITY_LIMIT: Duration = Duration::from_secs(30);
const GIT_CHANNEL_WALL_CLOCK_LIMIT: Duration = Duration::from_secs(120);
const MAX_GIT_CHANNEL_BYTES: u64 = 257 * 1024 * 1024;
const REPOSITORY_CREATE_USAGE: &str = "repo create NAME [--output human|json]";
const ISSUE_LIST_USAGE: &str = "issue list OWNER/REPOSITORY [--output human|json]";
const ISSUE_CREATE_USAGE: &str = "issue create OWNER/REPOSITORY [--output human|json]";
const ISSUE_COMMENT_USAGE: &str = "issue comment OWNER/REPOSITORY NUMBER [--output human|json]";
const ISSUE_CLOSE_USAGE: &str = "issue close OWNER/REPOSITORY NUMBER [--output human|json]";
const ISSUE_REOPEN_USAGE: &str = "issue reopen OWNER/REPOSITORY NUMBER [--output human|json]";
const PULL_REQUEST_LIST_USAGE: &str =
"pr list OWNER/REPOSITORY [--state open|closed|merged|all] [--output human|json]";
const PULL_REQUEST_CREATE_USAGE: &str =
"pr create OWNER/REPOSITORY BASE HEAD [--output human|json]";
const PULL_REQUEST_CLOSE_USAGE: &str = "pr close OWNER/REPOSITORY NUMBER [--output human|json]";
const PULL_REQUEST_REOPEN_USAGE: &str = "pr reopen OWNER/REPOSITORY NUMBER [--output human|json]";
const PULL_REQUEST_CHECKOUT_USAGE: &str =
"pr checkout OWNER/REPOSITORY NUMBER [--output human|json]";
const HELP_TEXT: &str = "\
Available tit SSH commands:
help
tit --version
auth ONE-TIME-SECRET
repo create NAME [--output human|json]
issue list OWNER/REPOSITORY [--output human|json]
issue create OWNER/REPOSITORY [--output human|json]
issue comment OWNER/REPOSITORY NUMBER [--output human|json]
issue close OWNER/REPOSITORY NUMBER [--output human|json]
issue reopen OWNER/REPOSITORY NUMBER [--output human|json]
pr list OWNER/REPOSITORY [--state open|closed|merged|all] [--output human|json]
pr create OWNER/REPOSITORY BASE HEAD [--output human|json]
pr close OWNER/REPOSITORY NUMBER [--output human|json]
pr reopen OWNER/REPOSITORY NUMBER [--output human|json]
pr checkout OWNER/REPOSITORY NUMBER [--output human|json]
Git clients can also use git-upload-pack and git-receive-pack.
";
const HELP_GUIDANCE: &str = "Use 'help' to list the available commands.";
pub(crate) type LoginApprover =
Arc<dyn Fn(&str, &str, &str) -> Result<(String, String), ()> + Send + Sync>;
struct SshRuntime {
max_connections: usize,
telemetry: Telemetry,
login: Option<LoginApprover>,
}
fn parse_auth_command(command: &[u8]) -> Option<String> {
let secret = command.strip_prefix(AUTH_COMMAND_PREFIX)?;
if secret.len() != 64
|| !secret
.iter()
.all(|byte| byte.is_ascii_digit() || matches!(byte, b'a'..=b'f'))
{
return None;
}
String::from_utf8(secret.to_vec()).ok()
}
pub(crate) struct RunningSshServer {
address: SocketAddr,
handle: russh::server::RunningServerHandle,
task: JoinHandle<std::io::Result<()>>,
audit: Arc<RequestAudit>,
}
#[derive(Clone)]
pub(crate) struct AuthorizedSshKeys {
keys: Arc<RwLock<HashMap<PublicKey, SshIdentity>>>,
}
#[derive(Clone, PartialEq, Eq)]
struct SshIdentity {
username: String,
fingerprint: String,
}
impl AuthorizedSshKeys {
pub(crate) fn new(keys: &[SshPublicKey]) -> Self {
Self {
keys: Arc::new(RwLock::new(key_map(keys))),
}
}
pub(crate) fn for_accounts(keys: Vec<(String, SshPublicKey)>) -> Self {
Self {
keys: Arc::new(RwLock::new(account_key_map(keys))),
}
}
pub(crate) fn replace_accounts(&self, keys: Vec<(String, SshPublicKey)>) {
if let Ok(mut current) = self.keys.write() {
*current = account_key_map(keys);
}
}
fn identity(&self, key: &PublicKey) -> Option<SshIdentity> {
self.keys.read().ok()?.get(key).cloned()
}
fn contains(&self, key: &PublicKey) -> bool {
self.keys.read().is_ok_and(|keys| keys.contains_key(key))
}
}
fn key_map(keys: &[SshPublicKey]) -> HashMap<PublicKey, SshIdentity> {
keys.iter()
.map(|key| {
let fingerprint = key.fingerprint().to_owned();
(
key.public_key().clone(),
SshIdentity {
username: fingerprint.clone(),
fingerprint,
},
)
})
.collect()
}
fn account_key_map(keys: Vec<(String, SshPublicKey)>) -> HashMap<PublicKey, SshIdentity> {
keys.into_iter()
.map(|(username, key)| {
(
key.public_key().clone(),
SshIdentity {
username,
fingerprint: key.fingerprint().to_owned(),
},
)
})
.collect()
}
impl RunningSshServer {
pub(crate) async fn start(
address: SocketAddr,
authorized_keys: &[SshPublicKey],
) -> Result<Self, SshServerError> {
let host_key = PrivateKey::random(&mut rng(), Algorithm::Ed25519)?;
Self::start_inner(address, authorized_keys, &[], None, host_key).await
}
pub(crate) async fn start_with_git(
address: SocketAddr,
authorized_keys: &[SshPublicKey],
repositories: GitRepositories,
) -> Result<Self, SshServerError> {
let host_key = PrivateKey::random(&mut rng(), Algorithm::Ed25519)?;
Self::start_inner(address, authorized_keys, &[], Some(repositories), host_key).await
}
pub(crate) async fn start_with_git_and_host_key(
address: SocketAddr,
authorized_keys: &[SshPublicKey],
repositories: GitRepositories,
host_key: PrivateKey,
) -> Result<Self, SshServerError> {
Self::start_inner(address, authorized_keys, &[], Some(repositories), host_key).await
}
pub(crate) async fn start_with_dynamic_keys(
address: SocketAddr,
authorized_keys: AuthorizedSshKeys,
repositories: GitRepositories,
host_key: PrivateKey,
max_connections: usize,
telemetry: Telemetry,
login: LoginApprover,
) -> Result<Self, SshServerError> {
recover_pushes(&repositories).await?;
Self::start_inner_with_keys(
address,
authorized_keys,
&[],
Some(repositories),
host_key,
SshRuntime {
max_connections,
telemetry,
login: Some(login),
},
)
.await
}
pub(crate) async fn start_with_git_writes(
address: SocketAddr,
authorized_keys: &[SshPublicKey],
writable_keys: &[SshPublicKey],
repositories: GitRepositories,
) -> Result<Self, SshServerError> {
recover_pushes(&repositories).await?;
let host_key = PrivateKey::random(&mut rng(), Algorithm::Ed25519)?;
Self::start_inner(
address,
authorized_keys,
writable_keys,
Some(repositories),
host_key,
)
.await
}
async fn start_inner(
address: SocketAddr,
authorized_keys: &[SshPublicKey],
writable_keys: &[SshPublicKey],
repositories: Option<GitRepositories>,
host_key: PrivateKey,
) -> Result<Self, SshServerError> {
Self::start_inner_with_keys(
address,
AuthorizedSshKeys::new(authorized_keys),
writable_keys,
repositories,
host_key,
SshRuntime {
max_connections: 1024,
telemetry: Telemetry::default(),
login: None,
},
)
.await
}
async fn start_inner_with_keys(
address: SocketAddr,
authorized_keys: AuthorizedSshKeys,
writable_keys: &[SshPublicKey],
repositories: Option<GitRepositories>,
host_key: PrivateKey,
runtime: SshRuntime,
) -> Result<Self, SshServerError> {
let listener = TcpListener::bind(address).await?;
let address = listener.local_addr()?;
let mut methods = MethodSet::empty();
methods.push(MethodKind::PublicKey);
let config = Arc::new(russh::server::Config {
methods,
auth_rejection_time: Duration::from_millis(250),
auth_rejection_time_initial: Some(Duration::ZERO),
keys: vec![host_key],
preferred: Preferred {
key: Cow::Owned(vec![
Algorithm::Ed25519,
Algorithm::Ecdsa {
curve: EcdsaCurve::NistP256,
},
]),
..Preferred::default()
},
max_auth_attempts: 3,
inactivity_timeout: Some(Duration::from_secs(30)),
nodelay: true,
..Default::default()
});
let writable_keys = Arc::new(
writable_keys
.iter()
.map(|key| key.public_key().clone())
.collect(),
);
let audit = Arc::new(RequestAudit::default());
let server = SshServer {
authorized_keys,
writable_keys,
audit: Arc::clone(&audit),
repositories: repositories.map(Arc::new),
login: runtime.login,
connections: Arc::new(Semaphore::new(runtime.max_connections)),
git_channels: Arc::new(Semaphore::new(MAX_GIT_CHANNELS_GLOBAL)),
attempts: AttemptLimiter::new(
SSH_ATTEMPTS_PER_MINUTE,
Duration::from_secs(60),
MAX_SSH_CLIENTS,
),
telemetry: runtime.telemetry,
};
let (handle_sender, handle_receiver) = oneshot::channel();
let task = tokio::spawn(async move {
let mut server = server;
let running = server.run_on_socket(config, &listener);
let _ = handle_sender.send(running.handle());
running.await
});
let handle = handle_receiver.await.map_err(|_| SshServerError::Startup)?;
Ok(Self {
address,
handle,
task,
audit,
})
}
pub(crate) fn address(&self) -> SocketAddr {
self.address
}
pub(crate) fn audit(&self) -> RequestAuditSnapshot {
self.audit.snapshot()
}
pub(crate) async fn shutdown(self) -> Result<(), SshServerError> {
self.handle.shutdown("tit test shutdown".to_owned());
self.task.await.map_err(|_| SshServerError::Join)??;
Ok(())
}
pub(crate) async fn shutdown_bounded(
mut self,
limit: Duration,
) -> Result<bool, SshServerError> {
self.handle.shutdown("tit shutdown".to_owned());
match tokio::time::timeout(limit, &mut self.task).await {
Ok(result) => {
result.map_err(|_| SshServerError::Join)??;
Ok(true)
}
Err(_) => {
self.task.abort();
let _ = self.task.await;
Ok(false)
}
}
}
}
async fn recover_pushes(repositories: &GitRepositories) -> Result<(), SshServerError> {
let database = repositories
.push_database()
.ok_or_else(|| SshServerError::Recovery("push storage is not configured".to_owned()))?
.to_owned();
tokio::task::spawn_blocking(move || {
crate::git::receive_pack::recover_incomplete_pushes(&database)
})
.await
.map_err(|_| SshServerError::Join)?
.map_err(|error| SshServerError::Recovery(error.to_string()))
}
#[derive(Debug, Error)]
pub(crate) enum SshServerError {
#[error("SSH listener error: {0}")]
Io(#[from] std::io::Error),
#[error("SSH key error: {0}")]
Key(#[from] ssh_key::Error),
#[error("SSH server did not start")]
Startup,
#[error("SSH server task failed")]
Join,
#[error("cannot recover Git writes: {0}")]
Recovery(String),
}
#[derive(Clone)]
struct SshServer {
authorized_keys: AuthorizedSshKeys,
writable_keys: Arc<HashSet<PublicKey>>,
audit: Arc<RequestAudit>,
repositories: Option<Arc<GitRepositories>>,
login: Option<LoginApprover>,
connections: Arc<Semaphore>,
git_channels: Arc<Semaphore>,
attempts: AttemptLimiter<IpAddr>,
telemetry: Telemetry,
}
impl Server for SshServer {
type Handler = SshSession;
fn new_client(&mut self, peer_address: Option<SocketAddr>) -> Self::Handler {
let connection_id = format!("{:032x}", rand::random::<u128>());
self.telemetry.ssh_connection(&connection_id);
SshSession {
authorized_keys: self.authorized_keys.clone(),
writable_keys: Arc::clone(&self.writable_keys),
audit: Arc::clone(&self.audit),
repositories: self.repositories.clone(),
login: self.login.clone(),
protocol: ProtocolVersion::V0,
exec_channels: HashMap::new(),
authenticated_identity: None,
authenticated_key: None,
authenticated_writer: false,
peer_address: peer_address
.map(|address| address.ip())
.unwrap_or(IpAddr::V4(Ipv4Addr::LOCALHOST)),
attempts: self.attempts.clone(),
_connection_permit: self.connections.clone().try_acquire_owned().ok(),
git_channels: Arc::clone(&self.git_channels),
telemetry: self.telemetry.clone(),
connection_id,
}
}
}
struct SshSession {
authorized_keys: AuthorizedSshKeys,
writable_keys: Arc<HashSet<PublicKey>>,
audit: Arc<RequestAudit>,
repositories: Option<Arc<GitRepositories>>,
login: Option<LoginApprover>,
protocol: ProtocolVersion,
exec_channels: HashMap<ChannelId, ExecChannel>,
authenticated_identity: Option<SshIdentity>,
authenticated_key: Option<PublicKey>,
authenticated_writer: bool,
peer_address: IpAddr,
attempts: AttemptLimiter<IpAddr>,
_connection_permit: Option<OwnedSemaphorePermit>,
git_channels: Arc<Semaphore>,
telemetry: Telemetry,
connection_id: String,
}
enum ExecChannel {
Upload(Box<UploadChannel>),
Receive(Box<ReceiveChannel>),
IssueInput(IssueInputChannel),
PullRequestCreate(PullRequestCreateChannel),
}
enum IssueInputKind {
Create,
Comment(i64),
}
struct IssueInputChannel {
actor: String,
owner: String,
repository: String,
output: CommandOutput,
kind: IssueInputKind,
input: Vec<u8>,
}
struct PullRequestCreateChannel {
actor: String,
owner: String,
repository: String,
base: String,
head: String,
output: CommandOutput,
input: Vec<u8>,
}
struct UploadChannel {
service: UploadPack,
request: Vec<u8>,
started_at: Instant,
last_activity: Instant,
transferred: u64,
cancelled: Arc<AtomicBool>,
_global_permit: OwnedSemaphorePermit,
}
struct ReceiveChannel {
service: ReceivePack,
owner: String,
repository: String,
identity: SshIdentity,
public_key: PublicKey,
authorized_keys: AuthorizedSshKeys,
commands: Vec<u8>,
commands_complete: bool,
pack: tokio::fs::File,
pack_bytes: u64,
started_at: Instant,
last_activity: Instant,
transferred: u64,
_global_permit: OwnedSemaphorePermit,
}
impl UploadChannel {
fn expired(&self) -> bool {
git_channel_expired(self.started_at, self.last_activity)
}
}
impl Drop for UploadChannel {
fn drop(&mut self) {
self.cancelled.store(true, Ordering::Relaxed);
}
}
impl ReceiveChannel {
fn expired(&self) -> bool {
git_channel_expired(self.started_at, self.last_activity)
}
}
fn git_channel_expired(started_at: Instant, last_activity: Instant) -> bool {
started_at.elapsed() > GIT_CHANNEL_WALL_CLOCK_LIMIT
|| last_activity.elapsed() > GIT_CHANNEL_INACTIVITY_LIMIT
}
fn reserve_git_channel(
active_channels: usize,
global_channels: &Arc<Semaphore>,
) -> Option<OwnedSemaphorePermit> {
if active_channels >= MAX_GIT_CHANNELS_PER_CONNECTION {
return None;
}
global_channels.clone().try_acquire_owned().ok()
}
impl Handler for SshSession {
type Error = russh::Error;
async fn auth_publickey_offered(
&mut self,
_user: &str,
public_key: &PublicKey,
) -> Result<Auth, Self::Error> {
Ok(self.authorize(public_key))
}
async fn auth_publickey(
&mut self,
_user: &str,
public_key: &PublicKey,
) -> Result<Auth, Self::Error> {
if self._connection_permit.is_none() || !self.attempts.allow(self.peer_address) {
self.telemetry.ssh_auth(&self.connection_id, false);
return Ok(Auth::reject());
}
if let Some(identity) = self.authorized_keys.identity(public_key) {
self.authenticated_identity = Some(identity);
self.authenticated_key = Some(public_key.clone());
self.authenticated_writer = self.writable_keys.contains(public_key);
self.telemetry.ssh_auth(&self.connection_id, true);
Ok(Auth::Accept)
} else {
self.telemetry.ssh_auth(&self.connection_id, false);
Ok(Auth::reject())
}
}
async fn channel_open_session(
&mut self,
_channel: Channel<Msg>,
reply: ChannelOpenHandle,
_session: &mut Session,
) -> Result<(), Self::Error> {
reply.accept().await;
Ok(())
}
async fn channel_open_x11(
&mut self,
_channel: Channel<Msg>,
_originator_address: &str,
_originator_port: u32,
_reply: ChannelOpenHandle,
_session: &mut Session,
) -> Result<(), Self::Error> {
self.audit.rejected_forward.fetch_add(1, Ordering::Relaxed);
Ok(())
}
async fn channel_open_direct_tcpip(
&mut self,
_channel: Channel<Msg>,
_host_to_connect: &str,
_port_to_connect: u32,
_originator_address: &str,
_originator_port: u32,
_reply: ChannelOpenHandle,
_session: &mut Session,
) -> Result<(), Self::Error> {
self.audit.rejected_forward.fetch_add(1, Ordering::Relaxed);
Ok(())
}
async fn channel_open_direct_streamlocal(
&mut self,
_channel: Channel<Msg>,
_socket_path: &str,
_reply: ChannelOpenHandle,
_session: &mut Session,
) -> Result<(), Self::Error> {
self.audit.rejected_forward.fetch_add(1, Ordering::Relaxed);
Ok(())
}
async fn pty_request(
&mut self,
channel: ChannelId,
_term: &str,
_col_width: u32,
_row_height: u32,
_pix_width: u32,
_pix_height: u32,
_modes: &[(Pty, u32)],
session: &mut Session,
) -> Result<(), Self::Error> {
self.audit.rejected_pty.fetch_add(1, Ordering::Relaxed);
session.channel_failure(channel)?;
Ok(())
}
async fn x11_request(
&mut self,
channel: ChannelId,
_single_connection: bool,
_x11_auth_protocol: &str,
_x11_auth_cookie: &str,
_x11_screen_number: u32,
session: &mut Session,
) -> Result<(), Self::Error> {
self.audit.rejected_forward.fetch_add(1, Ordering::Relaxed);
session.channel_failure(channel)?;
Ok(())
}
async fn env_request(
&mut self,
channel: ChannelId,
variable_name: &str,
variable_value: &str,
session: &mut Session,
) -> Result<(), Self::Error> {
if variable_name == GIT_PROTOCOL_VARIABLE && valid_git_protocol(variable_value) {
self.protocol = match variable_value {
"version=0" => ProtocolVersion::V0,
"version=1" => ProtocolVersion::V1,
"version=2" => ProtocolVersion::V2,
_ => unreachable!("the Git protocol value was validated"),
};
self.audit.accepted_env.fetch_add(1, Ordering::Relaxed);
session.channel_success(channel)?;
} else {
self.audit.rejected_env.fetch_add(1, Ordering::Relaxed);
session.channel_failure(channel)?;
}
Ok(())
}
async fn shell_request(
&mut self,
channel: ChannelId,
session: &mut Session,
) -> Result<(), Self::Error> {
self.audit.rejected_shell.fetch_add(1, Ordering::Relaxed);
session.channel_failure(channel)?;
session.close(channel)?;
Ok(())
}
async fn exec_request(
&mut self,
channel: ChannelId,
command: &[u8],
session: &mut Session,
) -> Result<(), Self::Error> {
let operation_id = format!("{:032x}", rand::random::<u128>());
self.telemetry.ssh_operation(&operation_id);
if command == VERSION_COMMAND {
self.audit.accepted_exec.fetch_add(1, Ordering::Relaxed);
session.channel_success(channel)?;
session.data(
channel,
format!("tit {}\n", env!("CARGO_PKG_VERSION")).into_bytes(),
)?;
session.exit_status_request(channel, 0)?;
session.eof(channel)?;
session.close(channel)?;
} else if command == HELP_COMMAND {
self.audit.accepted_exec.fetch_add(1, Ordering::Relaxed);
session.channel_success(channel)?;
session.data(channel, HELP_TEXT.as_bytes())?;
finish_git_channel(channel, 0, session)?;
} else if command.starts_with(AUTH_COMMAND_PREFIX) {
self.audit.accepted_exec.fetch_add(1, Ordering::Relaxed);
session.channel_success(channel)?;
let result = match (
parse_auth_command(command),
self.active_identity(),
self.login.clone(),
) {
(Some(secret), Some(identity), Some(login)) => {
let username = identity.username;
let fingerprint = identity.fingerprint;
tokio::task::spawn_blocking(move || login(&secret, &username, &fingerprint))
.await
.ok()
.and_then(Result::ok)
}
_ => None,
};
match result {
Some((origin, username)) => {
session.data(
channel,
format!(
"Authentication approved.\nOrigin: {}\nAccount: {}\nReturn to your browser and select Continue.\n",
origin, username
)
.into_bytes(),
)?;
finish_git_channel(channel, 0, session)?;
}
None => {
session.extended_data(
channel,
1,
format!(
"tit: The authentication approval is invalid or expired.\n{HELP_GUIDANCE}\n"
)
.into_bytes(),
)?;
finish_git_channel(channel, 1, session)?;
}
}
} else if is_repository_command(command) {
self.audit.accepted_exec.fetch_add(1, Ordering::Relaxed);
session.channel_success(channel)?;
let machine_requested = requests_json(command);
let result = match parse_repository_command(command) {
Ok(command) => match self.active_identity() {
Some(identity) => {
run_repository_command(
self.repositories.clone(),
identity.username.clone(),
command,
)
.await
}
None => Err(RepositoryCommandError::Unavailable),
},
Err(()) => Err(RepositoryCommandError::Usage),
};
send_repository_command_result(channel, result, machine_requested, session)?;
} else if is_issue_command(command) {
self.audit.accepted_exec.fetch_add(1, Ordering::Relaxed);
session.channel_success(channel)?;
let machine_requested = requests_json(command);
match (parse_issue_command(command), self.active_identity()) {
(Ok(IssueCommand::List(command)), Some(identity)) => {
let result =
run_issue_list(self.repositories.clone(), identity.username, command).await;
send_issue_list_result(channel, result, machine_requested, session)?;
}
(Ok(IssueCommand::Create(command)), Some(identity)) => {
self.exec_channels.insert(
channel,
ExecChannel::IssueInput(IssueInputChannel {
actor: identity.username,
owner: command.owner,
repository: command.repository,
output: command.output,
kind: IssueInputKind::Create,
input: Vec::new(),
}),
);
}
(Ok(IssueCommand::Comment(command)), Some(identity)) => {
self.exec_channels.insert(
channel,
ExecChannel::IssueInput(IssueInputChannel {
actor: identity.username,
owner: command.owner,
repository: command.repository,
output: command.output,
kind: IssueInputKind::Comment(command.number),
input: Vec::new(),
}),
);
}
(Ok(IssueCommand::SetState(command)), Some(identity)) => {
let result =
run_issue_set_state(self.repositories.clone(), identity.username, command)
.await;
send_issue_mutation_result(channel, result, machine_requested, session)?;
}
(Ok(_), None) => send_issue_error(
channel,
IssueCommandError::Unavailable,
machine_requested,
session,
)?,
(Err(()), _) => send_issue_error(
channel,
IssueCommandError::Usage,
machine_requested,
session,
)?,
}
} else if is_pull_request_command(command) {
self.audit.accepted_exec.fetch_add(1, Ordering::Relaxed);
session.channel_success(channel)?;
let machine_requested = requests_json(command);
match (parse_pull_request_command(command), self.active_identity()) {
(Ok(PullRequestCommand::Checkout(command)), Some(identity)) => {
let result = run_pull_request_checkout(
self.repositories.clone(),
identity.username,
command,
)
.await;
send_pull_request_checkout_result(channel, result, machine_requested, session)?;
}
(Ok(PullRequestCommand::List(command)), Some(identity)) => {
let result = run_pull_request_list(
self.repositories.clone(),
identity.username,
command,
)
.await;
send_pull_request_list_result(channel, result, machine_requested, session)?;
}
(Ok(PullRequestCommand::Create(command)), Some(identity)) => {
self.exec_channels.insert(
channel,
ExecChannel::PullRequestCreate(PullRequestCreateChannel {
actor: identity.username,
owner: command.owner,
repository: command.repository,
base: command.base,
head: command.head,
output: command.output,
input: Vec::new(),
}),
);
}
(Ok(PullRequestCommand::SetState(command)), Some(identity)) => {
let result = run_pull_request_set_state(
self.repositories.clone(),
identity.username,
command,
)
.await;
send_pull_request_mutation_result(channel, result, machine_requested, session)?;
}
(Ok(_), None) => send_pull_request_error(
channel,
PullRequestCommandError::Unavailable,
machine_requested,
session,
)?,
(Err(()), _) => send_pull_request_error(
channel,
PullRequestCommandError::Usage,
machine_requested,
session,
)?,
}
} else {
let service = self.open_git_service(command).await;
if let Some(service) = service {
self.audit.accepted_exec.fetch_add(1, Ordering::Relaxed);
session.channel_success(channel)?;
match service {
InitialGitService::Upload {
service,
advertisement,
global_permit,
} => {
let started_at = Instant::now();
let advertisement_bytes =
u64::try_from(advertisement.len()).unwrap_or(u64::MAX);
session.data(channel, advertisement)?;
self.exec_channels.insert(
channel,
ExecChannel::Upload(Box::new(UploadChannel {
service: *service,
request: Vec::new(),
started_at,
last_activity: started_at,
transferred: advertisement_bytes,
cancelled: Arc::new(AtomicBool::new(false)),
_global_permit: global_permit,
})),
);
}
InitialGitService::Receive(receive) => {
let InitialReceiveService {
service,
advertisement,
owner,
repository,
identity,
public_key,
global_permit,
} = *receive;
let started_at = Instant::now();
let advertisement_bytes =
u64::try_from(advertisement.len()).unwrap_or(u64::MAX);
session.data(channel, advertisement)?;
let pack = tokio::fs::File::create(service.incoming_pack()).await;
match pack {
Ok(pack) => {
self.exec_channels.insert(
channel,
ExecChannel::Receive(Box::new(ReceiveChannel {
service: *service,
owner,
repository,
identity,
public_key,
authorized_keys: self.authorized_keys.clone(),
commands: Vec::new(),
commands_complete: false,
pack,
pack_bytes: 0,
started_at,
last_activity: started_at,
transferred: advertisement_bytes,
_global_permit: global_permit,
})),
);
}
Err(_) => fail_git_channel(channel, session)?,
}
}
}
} else {
self.audit.accepted_exec.fetch_add(1, Ordering::Relaxed);
session.channel_success(channel)?;
if requests_json(command) {
session.data(
channel,
json_line(serde_json::json!({
"version": 1,
"status": "error",
"error": { "code": "invalid-command" },
})),
)?;
} else {
session.extended_data(
channel,
1,
format!("tit: The command is not valid.\n{HELP_GUIDANCE}\n").into_bytes(),
)?;
}
finish_git_channel(channel, 1, session)?;
}
}
Ok(())
}
async fn data(
&mut self,
channel: ChannelId,
data: &[u8],
session: &mut Session,
) -> Result<(), Self::Error> {
let Some(exec) = self.exec_channels.remove(&channel) else {
return Ok(());
};
let mut git = match exec {
ExecChannel::IssueInput(mut issue) => {
if append_issue_input(&mut issue.input, data).is_err() {
send_issue_error(
channel,
IssueCommandError::Input,
issue.output == CommandOutput::Json,
session,
)?;
} else {
self.exec_channels
.insert(channel, ExecChannel::IssueInput(issue));
}
return Ok(());
}
ExecChannel::PullRequestCreate(mut pull_request) => {
if append_issue_input(&mut pull_request.input, data).is_err() {
send_pull_request_error(
channel,
PullRequestCommandError::Input,
pull_request.output == CommandOutput::Json,
session,
)?;
} else {
self.exec_channels
.insert(channel, ExecChannel::PullRequestCreate(pull_request));
}
return Ok(());
}
ExecChannel::Upload(git) => git,
ExecChannel::Receive(mut git) => {
if git.expired() {
fail_git_channel(channel, session)?;
return Ok(());
}
git.last_activity = Instant::now();
git.transferred = git
.transferred
.saturating_add(u64::try_from(data.len()).unwrap_or(u64::MAX));
if !git_channel_bytes_within_limit(git.transferred, 0) {
fail_git_channel(channel, session)?;
return Ok(());
}
if receive_data(&mut git, data).await.is_err() {
fail_git_channel(channel, session)?;
} else if git.commands_complete
&& matches!(git.service.expects_pack(&git.commands), Ok(false))
{
send_receive_result(
channel,
finish_receive(self.repositories.clone(), git).await,
session,
)?;
} else {
self.exec_channels
.insert(channel, ExecChannel::Receive(git));
}
return Ok(());
}
};
if git.expired() {
fail_git_channel(channel, session)?;
return Ok(());
}
git.last_activity = Instant::now();
let bytes = u64::try_from(data.len()).unwrap_or(u64::MAX);
git.transferred = git.transferred.saturating_add(bytes);
if !git_channel_bytes_within_limit(git.transferred, 0) {
fail_git_channel(channel, session)?;
return Ok(());
}
if git.request.len().saturating_add(data.len()) > MAX_REQUEST_BYTES {
fail_git_channel(channel, session)?;
return Ok(());
}
git.request.extend_from_slice(data);
let packets = match decode(&git.request) {
Ok(packets) => packets,
Err(super::git::packetline::PacketLineError::TruncatedHeader)
| Err(super::git::packetline::PacketLineError::TruncatedPacket) => {
self.exec_channels.insert(channel, ExecChannel::Upload(git));
return Ok(());
}
Err(_) => {
fail_git_channel(channel, session)?;
return Ok(());
}
};
if packets == [Packet::Flush] {
finish_git_channel(channel, 0, session)?;
return Ok(());
}
match self.protocol {
ProtocolVersion::V0 | ProtocolVersion::V1 => {
let done = packets.last().is_some_and(
|packet| matches!(packet, Packet::Data(line) if trim_line(line) == b"done"),
);
if done {
match respond_git(self.repositories.clone(), self.protocol, git).await {
Some((git, Ok(response))) => {
if !git_response_within_limit(&git, response.len()) {
fail_git_channel(channel, session)?;
return Ok(());
}
session.data(channel, response)?;
finish_git_channel(channel, 0, session)?;
}
Some((_, Err(_))) | None => fail_git_channel(channel, session)?,
}
} else {
self.exec_channels.insert(channel, ExecChannel::Upload(git));
}
}
ProtocolVersion::V2 => {
if packets.last() != Some(&Packet::Flush) {
self.exec_channels.insert(channel, ExecChannel::Upload(git));
return Ok(());
}
let fetch = packets.iter().any(
|packet| matches!(packet, Packet::Data(line) if trim_line(line) == b"command=fetch"),
);
let done = packets.iter().any(
|packet| matches!(packet, Packet::Data(line) if trim_line(line) == b"done"),
);
match respond_git(self.repositories.clone(), self.protocol, git).await {
Some((mut git, Ok(response))) => {
if !git_response_within_limit(&git, response.len()) {
fail_git_channel(channel, session)?;
return Ok(());
}
git.transferred = git
.transferred
.saturating_add(u64::try_from(response.len()).unwrap_or(u64::MAX));
session.data(channel, response)?;
if fetch && done {
finish_git_channel(channel, 0, session)?;
} else {
git.request.clear();
self.exec_channels.insert(channel, ExecChannel::Upload(git));
}
}
Some((_, Err(_))) | None => fail_git_channel(channel, session)?,
}
}
}
Ok(())
}
async fn channel_close(
&mut self,
channel: ChannelId,
_session: &mut Session,
) -> Result<(), Self::Error> {
self.exec_channels.remove(&channel);
Ok(())
}
async fn channel_eof(
&mut self,
channel: ChannelId,
session: &mut Session,
) -> Result<(), Self::Error> {
match self.exec_channels.remove(&channel) {
Some(ExecChannel::Receive(git)) => {
if !git.commands_complete || git.expired() {
fail_git_channel(channel, session)?;
return Ok(());
}
let result = finish_receive(self.repositories.clone(), git).await;
send_receive_result(channel, result, session)?;
}
Some(ExecChannel::IssueInput(issue)) => {
let active = self
.active_identity()
.is_some_and(|identity| identity.username == issue.actor);
let machine_requested = issue.output == CommandOutput::Json;
let result = if active {
run_issue_input(self.repositories.clone(), issue).await
} else {
Err(IssueCommandError::Unavailable)
};
send_issue_mutation_result(channel, result, machine_requested, session)?;
}
Some(ExecChannel::PullRequestCreate(pull_request)) => {
let active = self
.active_identity()
.is_some_and(|identity| identity.username == pull_request.actor);
let machine_requested = pull_request.output == CommandOutput::Json;
let result = if active {
run_pull_request_create(self.repositories.clone(), pull_request).await
} else {
Err(PullRequestCommandError::Unavailable)
};
send_pull_request_mutation_result(channel, result, machine_requested, session)?;
}
Some(ExecChannel::Upload(_)) | None => {}
}
Ok(())
}
async fn subsystem_request(
&mut self,
channel: ChannelId,
_name: &str,
session: &mut Session,
) -> Result<(), Self::Error> {
self.audit.rejected_exec.fetch_add(1, Ordering::Relaxed);
session.channel_failure(channel)?;
session.close(channel)?;
Ok(())
}
async fn agent_request(
&mut self,
channel: ChannelId,
session: &mut Session,
) -> Result<bool, Self::Error> {
self.audit.rejected_agent.fetch_add(1, Ordering::Relaxed);
session.channel_failure(channel)?;
Ok(false)
}
async fn tcpip_forward(
&mut self,
_address: &str,
_port: &mut u32,
_session: &mut Session,
) -> Result<bool, Self::Error> {
self.audit.rejected_forward.fetch_add(1, Ordering::Relaxed);
Ok(false)
}
async fn streamlocal_forward(
&mut self,
_socket_path: &str,
_session: &mut Session,
) -> Result<bool, Self::Error> {
self.audit.rejected_forward.fetch_add(1, Ordering::Relaxed);
Ok(false)
}
}
#[derive(Clone, Copy, Eq, PartialEq)]
enum CommandOutput {
Human,
Json,
}
struct RepositoryCreateCommand {
slug: String,
output: CommandOutput,
}
fn is_repository_command(command: &[u8]) -> bool {
command == b"repo" || command.starts_with(b"repo ")
}
fn requests_json(command: &[u8]) -> bool {
std::str::from_utf8(command).is_ok_and(|command| {
let tokens = command.split_ascii_whitespace().collect::<Vec<_>>();
tokens
.windows(2)
.any(|tokens| tokens == ["--output", "json"])
})
}
fn parse_repository_command(command: &[u8]) -> Result<RepositoryCreateCommand, ()> {
if command.len() > MAX_REPOSITORY_COMMAND_BYTES || !command.is_ascii() {
return Err(());
}
let command = std::str::from_utf8(command).map_err(|_| ())?;
if command
.bytes()
.any(|byte| byte.is_ascii_control() && byte != b' ')
{
return Err(());
}
let mut tokens = command.split_ascii_whitespace();
if tokens.next() != Some("repo") || tokens.next() != Some("create") {
return Err(());
}
let slug = tokens.next().ok_or(())?.to_owned();
let mut output = None;
while let Some(option) = tokens.next() {
let value = tokens.next().ok_or(())?;
match option {
"--output" if output.is_none() => {
output = Some(match value {
"human" => CommandOutput::Human,
"json" => CommandOutput::Json,
_ => return Err(()),
});
}
_ => return Err(()),
}
}
Ok(RepositoryCreateCommand {
slug,
output: output.unwrap_or(CommandOutput::Human),
})
}
enum RepositoryCommandError {
Usage,
Unavailable,
Create(RepositoryServiceError),
}
async fn run_repository_command(
repositories: Option<Arc<GitRepositories>>,
actor: String,
command: RepositoryCreateCommand,
) -> Result<(crate::store::RepositoryRecord, CommandOutput), RepositoryCommandError> {
let repositories = repositories.ok_or(RepositoryCommandError::Unavailable)?;
let database = repositories
.push_database()
.ok_or(RepositoryCommandError::Unavailable)?
.to_owned();
let root = repositories.repository_root().to_owned();
let permit = repositories
.blocking_permit()
.await
.map_err(|_| RepositoryCommandError::Unavailable)?;
let output = command.output;
let correlation_id = format!("{:032x}", rand::random::<u128>());
tokio::task::spawn_blocking(move || {
let _permit = permit;
RepositoryService::new(&database, &root)
.create_for_account(
&actor,
&command.slug,
gix::hash::Kind::Sha1,
&correlation_id,
)
.map(|repository| (repository, output))
.map_err(RepositoryCommandError::Create)
})
.await
.map_err(|_| RepositoryCommandError::Unavailable)?
}
fn send_repository_command_result(
channel: ChannelId,
result: Result<(crate::store::RepositoryRecord, CommandOutput), RepositoryCommandError>,
machine_requested: bool,
session: &mut Session,
) -> Result<(), russh::Error> {
match result {
Ok((repository, CommandOutput::Human)) => {
session.data(
channel,
format!(
"Created repository {}/{}.\n",
repository.owner, repository.slug
)
.into_bytes(),
)?;
finish_git_channel(channel, 0, session)
}
Ok((repository, CommandOutput::Json)) => {
session.data(
channel,
format!(
"{{\"version\":1,\"status\":\"success\",\"repository\":{{\"owner\":\"{}\",\"name\":\"{}\"}}}}\n",
repository.owner, repository.slug
)
.into_bytes(),
)?;
finish_git_channel(channel, 0, session)
}
Err(error) => {
if machine_requested {
session.data(
channel,
format!(
"{{\"version\":1,\"status\":\"error\",\"error\":{{\"code\":\"{}\"}}}}\n",
repository_command_error_code(&error)
)
.into_bytes(),
)?;
} else {
session.extended_data(
channel,
1,
format!("tit: {}\n", repository_command_error_message(&error)).into_bytes(),
)?;
}
finish_git_channel(channel, 1, session)
}
}
}
fn repository_command_error_code(error: &RepositoryCommandError) -> &'static str {
match error {
RepositoryCommandError::Usage => "invalid-command",
RepositoryCommandError::Unavailable => "service-unavailable",
RepositoryCommandError::Create(RepositoryServiceError::Store(
crate::store::StoreError::RepositoryExists(_, _),
)) => "repository-exists",
RepositoryCommandError::Create(RepositoryServiceError::Auth(_))
| RepositoryCommandError::Create(RepositoryServiceError::RepositoryName(_)) => {
"invalid-name"
}
RepositoryCommandError::Create(RepositoryServiceError::Store(
crate::store::StoreError::AccountNotFound(_),
)) => "account-unavailable",
RepositoryCommandError::Create(_) => "repository-create-failed",
}
}
fn repository_command_error_message(error: &RepositoryCommandError) -> String {
match repository_command_error_code(error) {
"invalid-command" => {
format!("usage: {REPOSITORY_CREATE_USAGE}\n{HELP_GUIDANCE}")
}
"repository-exists" => "A repository with this name already exists.".to_owned(),
"invalid-name" => "The repository name is not valid.".to_owned(),
"account-unavailable" => "The account is not active.".to_owned(),
"service-unavailable" => "The repository service is not available.".to_owned(),
_ => "The repository could not be created.".to_owned(),
}
}
enum IssueCommand {
List(IssueListCommand),
Create(IssueCreateCommand),
Comment(IssueNumberCommand),
SetState(IssueStateCommand),
}
struct IssueListCommand {
owner: String,
repository: String,
output: CommandOutput,
}
struct IssueCreateCommand {
owner: String,
repository: String,
output: CommandOutput,
}
struct IssueNumberCommand {
owner: String,
repository: String,
number: i64,
output: CommandOutput,
}
struct IssueStateCommand {
owner: String,
repository: String,
number: i64,
state: &'static str,
output: CommandOutput,
}
struct IssueListResult {
repository: crate::store::RepositoryRecord,
issues: Vec<crate::store::IssueRecord>,
output: CommandOutput,
}
struct IssueMutationResult {
owner: String,
repository: String,
number: i64,
operation: &'static str,
issue: Option<crate::store::IssueRecord>,
comment_id: Option<String>,
output: CommandOutput,
}
enum IssueCommandError {
Usage,
Input,
Unavailable,
Service(IssueError),
}
fn is_issue_command(command: &[u8]) -> bool {
command == b"issue" || command.starts_with(b"issue ")
}
fn parse_issue_command(command: &[u8]) -> Result<IssueCommand, ()> {
if command.len() > MAX_ISSUE_COMMAND_BYTES || !command.is_ascii() {
return Err(());
}
let command = std::str::from_utf8(command).map_err(|_| ())?;
if command
.bytes()
.any(|byte| byte.is_ascii_control() && byte != b' ')
{
return Err(());
}
let mut tokens = command.split_ascii_whitespace();
if tokens.next() != Some("issue") {
return Err(());
}
let operation = tokens.next().ok_or(())?;
let target = tokens.next().ok_or(())?;
let (owner, repository) = target.split_once('/').ok_or(())?;
if owner.is_empty() || repository.is_empty() || repository.contains('/') {
return Err(());
}
match operation {
"list" | "create" => {
let output = parse_output_options(tokens)?;
if operation == "list" {
Ok(IssueCommand::List(IssueListCommand {
owner: owner.to_owned(),
repository: repository.to_owned(),
output,
}))
} else {
Ok(IssueCommand::Create(IssueCreateCommand {
owner: owner.to_owned(),
repository: repository.to_owned(),
output,
}))
}
}
"comment" | "close" | "reopen" => {
let number = tokens.next().ok_or(())?.parse::<i64>().map_err(|_| ())?;
if number < 1 {
return Err(());
}
let output = parse_output_options(tokens)?;
if operation == "comment" {
Ok(IssueCommand::Comment(IssueNumberCommand {
owner: owner.to_owned(),
repository: repository.to_owned(),
number,
output,
}))
} else {
Ok(IssueCommand::SetState(IssueStateCommand {
owner: owner.to_owned(),
repository: repository.to_owned(),
number,
state: if operation == "close" {
"closed"
} else {
"open"
},
output,
}))
}
}
_ => Err(()),
}
}
fn parse_output_options<'a>(
mut tokens: impl Iterator<Item = &'a str>,
) -> Result<CommandOutput, ()> {
let mut output = None;
while let Some(option) = tokens.next() {
let value = tokens.next().ok_or(())?;
if option != "--output" || output.is_some() {
return Err(());
}
output = Some(match value {
"human" => CommandOutput::Human,
"json" => CommandOutput::Json,
_ => return Err(()),
});
}
Ok(output.unwrap_or(CommandOutput::Human))
}
async fn run_issue_list(
repositories: Option<Arc<GitRepositories>>,
actor: String,
command: IssueListCommand,
) -> Result<IssueListResult, IssueCommandError> {
let repositories = repositories.ok_or(IssueCommandError::Unavailable)?;
let database = repositories
.push_database()
.ok_or(IssueCommandError::Unavailable)?
.to_owned();
let permit = repositories
.blocking_permit()
.await
.map_err(|_| IssueCommandError::Unavailable)?;
tokio::task::spawn_blocking(move || {
let _permit = permit;
IssueService::new(&database)
.list(&command.owner, &command.repository, Some(&actor))
.map(|(repository, issues)| IssueListResult {
repository,
issues,
output: command.output,
})
.map_err(IssueCommandError::Service)
})
.await
.map_err(|_| IssueCommandError::Unavailable)?
}
async fn run_issue_input(
repositories: Option<Arc<GitRepositories>>,
command: IssueInputChannel,
) -> Result<IssueMutationResult, IssueCommandError> {
let repositories = repositories.ok_or(IssueCommandError::Unavailable)?;
let database = repositories
.push_database()
.ok_or(IssueCommandError::Unavailable)?
.to_owned();
let permit = repositories
.blocking_permit()
.await
.map_err(|_| IssueCommandError::Unavailable)?;
tokio::task::spawn_blocking(move || {
let _permit = permit;
let service = IssueService::new(&database);
match command.kind {
IssueInputKind::Create => {
let (title, body) = parse_issue_input(&command.input)?;
service
.create(
&command.owner,
&command.repository,
&command.actor,
&title,
&body,
)
.map(|issue| IssueMutationResult {
owner: command.owner,
repository: command.repository,
number: issue.number,
operation: "created",
issue: Some(issue),
comment_id: None,
output: command.output,
})
.map_err(IssueCommandError::Service)
}
IssueInputKind::Comment(number) => {
let body =
std::str::from_utf8(&command.input).map_err(|_| IssueCommandError::Input)?;
service
.comment(
&command.owner,
&command.repository,
number,
&command.actor,
body,
)
.map(|comment_id| IssueMutationResult {
owner: command.owner,
repository: command.repository,
number,
operation: "commented",
issue: None,
comment_id: Some(comment_id),
output: command.output,
})
.map_err(IssueCommandError::Service)
}
}
})
.await
.map_err(|_| IssueCommandError::Unavailable)?
}
async fn run_issue_set_state(
repositories: Option<Arc<GitRepositories>>,
actor: String,
command: IssueStateCommand,
) -> Result<IssueMutationResult, IssueCommandError> {
let repositories = repositories.ok_or(IssueCommandError::Unavailable)?;
let database = repositories
.push_database()
.ok_or(IssueCommandError::Unavailable)?
.to_owned();
let permit = repositories
.blocking_permit()
.await
.map_err(|_| IssueCommandError::Unavailable)?;
tokio::task::spawn_blocking(move || {
let _permit = permit;
IssueService::new(&database)
.set_state(
&command.owner,
&command.repository,
command.number,
&actor,
command.state,
)
.map(|()| IssueMutationResult {
owner: command.owner,
repository: command.repository,
number: command.number,
operation: if command.state == "closed" {
"closed"
} else {
"reopened"
},
issue: None,
comment_id: None,
output: command.output,
})
.map_err(IssueCommandError::Service)
})
.await
.map_err(|_| IssueCommandError::Unavailable)?
}
fn parse_issue_input(input: &[u8]) -> Result<(String, String), IssueCommandError> {
let (title, body) = input
.iter()
.position(|byte| *byte == b'\n')
.map_or((input, &[][..]), |end| (&input[..end], &input[end + 1..]));
let title = title.strip_suffix(b"\r").unwrap_or(title);
Ok((
std::str::from_utf8(title)
.map_err(|_| IssueCommandError::Input)?
.to_owned(),
std::str::from_utf8(body)
.map_err(|_| IssueCommandError::Input)?
.to_owned(),
))
}
fn append_issue_input(input: &mut Vec<u8>, data: &[u8]) -> Result<(), ()> {
if input.len().saturating_add(data.len()) > MAX_ISSUE_INPUT_BYTES {
return Err(());
}
input.extend_from_slice(data);
Ok(())
}
fn send_issue_list_result(
channel: ChannelId,
result: Result<IssueListResult, IssueCommandError>,
machine_requested: bool,
session: &mut Session,
) -> Result<(), russh::Error> {
match result {
Ok(result) => {
let data = match result.output {
CommandOutput::Human => issue_list_human(&result),
CommandOutput::Json => issue_list_json(&result),
};
session.data(channel, data)?;
finish_git_channel(channel, 0, session)
}
Err(error) => send_issue_error(channel, error, machine_requested, session),
}
}
fn issue_list_human(result: &IssueListResult) -> Vec<u8> {
if result.issues.is_empty() {
return b"No issues.\n".to_vec();
}
let mut output = String::new();
for issue in &result.issues {
output.push_str(&format!(
"#{} {} {}\n",
issue.number, issue.state, issue.title
));
}
output.into_bytes()
}
fn issue_list_json(result: &IssueListResult) -> Vec<u8> {
let issues = result
.issues
.iter()
.map(|issue| {
serde_json::json!({
"number": issue.number,
"title": issue.title,
"state": issue.state,
"author": issue.author,
"created_at": issue.created_at,
"updated_at": issue.updated_at,
})
})
.collect::<Vec<_>>();
json_line(serde_json::json!({
"version": 1,
"status": "success",
"repository": {
"owner": result.repository.owner,
"name": result.repository.slug,
},
"issues": issues,
}))
}
fn send_issue_mutation_result(
channel: ChannelId,
result: Result<IssueMutationResult, IssueCommandError>,
machine_requested: bool,
session: &mut Session,
) -> Result<(), russh::Error> {
match result {
Ok(result) => {
let data = match result.output {
CommandOutput::Human => format!(
"{} issue {}/{}#{}.\n",
sentence_case(result.operation),
result.owner,
result.repository,
result.number
)
.into_bytes(),
CommandOutput::Json => json_line(serde_json::json!({
"version": 1,
"status": "success",
"repository": {
"owner": result.owner,
"name": result.repository,
},
"issue": {
"number": result.number,
"title": result.issue.as_ref().map(|issue| &issue.title),
"state": result.issue.as_ref().map(|issue| &issue.state),
"author": result.issue.as_ref().map(|issue| &issue.author),
"created_at": result.issue.as_ref().map(|issue| issue.created_at),
"updated_at": result.issue.as_ref().map(|issue| issue.updated_at),
},
"operation": result.operation,
"comment_id": result.comment_id,
})),
};
session.data(channel, data)?;
finish_git_channel(channel, 0, session)
}
Err(error) => send_issue_error(channel, error, machine_requested, session),
}
}
fn sentence_case(value: &str) -> String {
let mut characters = value.chars();
match characters.next() {
Some(first) => first.to_uppercase().chain(characters).collect(),
None => String::new(),
}
}
fn send_issue_error(
channel: ChannelId,
error: IssueCommandError,
machine_requested: bool,
session: &mut Session,
) -> Result<(), russh::Error> {
if machine_requested {
session.data(
channel,
json_line(serde_json::json!({
"version": 1,
"status": "error",
"error": { "code": issue_command_error_code(&error) },
})),
)?;
} else {
session.extended_data(
channel,
1,
format!("tit: {}\n", issue_command_error_message(&error)).into_bytes(),
)?;
}
finish_git_channel(channel, 1, session)
}
fn json_line(value: serde_json::Value) -> Vec<u8> {
let mut output = serde_json::to_vec(&value).expect("a JSON value can be serialized");
output.push(b'\n');
output
}
fn issue_command_error_code(error: &IssueCommandError) -> &'static str {
match error {
IssueCommandError::Usage => "invalid-command",
IssueCommandError::Input
| IssueCommandError::Service(IssueError::Title | IssueError::Body) => "invalid-input",
IssueCommandError::Service(IssueError::Auth(_) | IssueError::RepositoryName(_)) => {
"invalid-target"
}
IssueCommandError::Service(IssueError::Store(
crate::store::StoreError::RepositoryNotFound(_, _)
| crate::store::StoreError::IssueHidden,
)) => "repository-unavailable",
IssueCommandError::Service(IssueError::Store(crate::store::StoreError::IssueDenied)) => {
"permission-denied"
}
IssueCommandError::Service(IssueError::Store(
crate::store::StoreError::AccountNotFound(_),
)) => "account-unavailable",
IssueCommandError::Unavailable => "service-unavailable",
IssueCommandError::Service(_) => "issue-command-failed",
}
}
fn issue_command_error_message(error: &IssueCommandError) -> String {
match issue_command_error_code(error) {
"invalid-command" => {
format!(
"usage: {ISSUE_LIST_USAGE}, {ISSUE_CREATE_USAGE}, {ISSUE_COMMENT_USAGE}, \
{ISSUE_CLOSE_USAGE}, or {ISSUE_REOPEN_USAGE}\n{HELP_GUIDANCE}"
)
}
"invalid-input" => {
"The first input line must be a valid title. The remaining input is the body."
.to_owned()
}
"invalid-target" => "The repository name is not valid.".to_owned(),
"repository-unavailable" => "The repository is not available.".to_owned(),
"permission-denied" => "The account cannot create an issue in this repository.".to_owned(),
"account-unavailable" => "The account is not active.".to_owned(),
"service-unavailable" => "The issue service is not available.".to_owned(),
_ => "The issue command could not be completed.".to_owned(),
}
}
struct PullRequestCheckoutCommand {
owner: String,
repository: String,
number: i64,
output: CommandOutput,
}
enum PullRequestCommand {
Checkout(PullRequestCheckoutCommand),
List(PullRequestListCommand),
Create(PullRequestCreateCommand),
SetState(PullRequestStateCommand),
}
struct PullRequestListCommand {
owner: String,
repository: String,
state: String,
output: CommandOutput,
}
struct PullRequestCreateCommand {
owner: String,
repository: String,
base: String,
head: String,
output: CommandOutput,
}
struct PullRequestStateCommand {
owner: String,
repository: String,
number: i64,
state: &'static str,
output: CommandOutput,
}
struct PullRequestCheckoutResult {
owner: String,
repository: String,
number: i64,
output: CommandOutput,
}
struct PullRequestListResult {
repository: crate::store::RepositoryRecord,
pull_requests: Vec<crate::store::PullRequestRecord>,
state: String,
output: CommandOutput,
}
struct PullRequestMutationResult {
owner: String,
repository: String,
number: i64,
operation: &'static str,
pull_request: Option<crate::store::PullRequestRecord>,
output: CommandOutput,
}
enum PullRequestCommandError {
Usage,
Input,
Unavailable,
Store(StoreError),
Service(PullRequestError),
}
fn is_pull_request_command(command: &[u8]) -> bool {
command == b"pr" || command.starts_with(b"pr ")
}
fn parse_pull_request_command(command: &[u8]) -> Result<PullRequestCommand, ()> {
if command.len() > MAX_PULL_REQUEST_COMMAND_BYTES || !command.is_ascii() {
return Err(());
}
let command = std::str::from_utf8(command).map_err(|_| ())?;
if command
.bytes()
.any(|byte| byte.is_ascii_control() && byte != b' ')
{
return Err(());
}
let mut tokens = command.split_ascii_whitespace();
if tokens.next() != Some("pr") {
return Err(());
}
let operation = tokens.next().ok_or(())?;
let target = tokens.next().ok_or(())?;
let (owner, repository) = target.split_once('/').ok_or(())?;
if owner.is_empty() || repository.is_empty() || repository.contains('/') {
return Err(());
}
match operation {
"list" => {
let (state, output) = parse_pull_request_list_options(tokens)?;
Ok(PullRequestCommand::List(PullRequestListCommand {
owner: owner.to_owned(),
repository: repository.to_owned(),
state,
output,
}))
}
"create" => {
let base = tokens.next().ok_or(())?.to_owned();
let head = tokens.next().ok_or(())?.to_owned();
let output = parse_output_options(tokens)?;
Ok(PullRequestCommand::Create(PullRequestCreateCommand {
owner: owner.to_owned(),
repository: repository.to_owned(),
base,
head,
output,
}))
}
"checkout" | "close" | "reopen" => {
let number = tokens.next().ok_or(())?.parse::<i64>().map_err(|_| ())?;
if number < 1 {
return Err(());
}
let output = parse_output_options(tokens)?;
if operation == "checkout" {
Ok(PullRequestCommand::Checkout(PullRequestCheckoutCommand {
owner: owner.to_owned(),
repository: repository.to_owned(),
number,
output,
}))
} else {
Ok(PullRequestCommand::SetState(PullRequestStateCommand {
owner: owner.to_owned(),
repository: repository.to_owned(),
number,
state: if operation == "close" {
"closed"
} else {
"open"
},
output,
}))
}
}
_ => Err(()),
}
}
fn parse_pull_request_list_options<'a>(
mut tokens: impl Iterator<Item = &'a str>,
) -> Result<(String, CommandOutput), ()> {
let mut state = None;
let mut output = None;
while let Some(option) = tokens.next() {
let value = tokens.next().ok_or(())?;
match option {
"--state" if state.is_none() => {
if !matches!(value, "open" | "closed" | "merged" | "all") {
return Err(());
}
state = Some(value.to_owned());
}
"--output" if output.is_none() => {
output = Some(match value {
"human" => CommandOutput::Human,
"json" => CommandOutput::Json,
_ => return Err(()),
});
}
_ => return Err(()),
}
}
Ok((
state.unwrap_or_else(|| "open".to_owned()),
output.unwrap_or(CommandOutput::Human),
))
}
async fn run_pull_request_checkout(
repositories: Option<Arc<GitRepositories>>,
actor: String,
command: PullRequestCheckoutCommand,
) -> Result<PullRequestCheckoutResult, PullRequestCommandError> {
let repositories = repositories.ok_or(PullRequestCommandError::Unavailable)?;
let database = repositories
.push_database()
.ok_or(PullRequestCommandError::Unavailable)?
.to_owned();
let permit = repositories
.blocking_permit()
.await
.map_err(|_| PullRequestCommandError::Unavailable)?;
tokio::task::spawn_blocking(move || {
let _permit = permit;
Store::open(&database)
.and_then(|store| {
store.pull_request(
&command.owner,
&command.repository,
command.number,
Some(&actor),
)
})
.map(|_| PullRequestCheckoutResult {
owner: command.owner,
repository: command.repository,
number: command.number,
output: command.output,
})
.map_err(PullRequestCommandError::Store)
})
.await
.map_err(|_| PullRequestCommandError::Unavailable)?
}
async fn run_pull_request_list(
repositories: Option<Arc<GitRepositories>>,
actor: String,
command: PullRequestListCommand,
) -> Result<PullRequestListResult, PullRequestCommandError> {
let repositories = repositories.ok_or(PullRequestCommandError::Unavailable)?;
let database = repositories
.push_database()
.ok_or(PullRequestCommandError::Unavailable)?
.to_owned();
let root = repositories.repository_root().to_owned();
let permit = repositories
.blocking_permit()
.await
.map_err(|_| PullRequestCommandError::Unavailable)?;
tokio::task::spawn_blocking(move || {
let _permit = permit;
PullRequestService::new(&database, &root)
.list_page(
&command.owner,
&command.repository,
Some(&actor),
&command.state,
1,
)
.map(|(repository, page, _)| PullRequestListResult {
repository,
pull_requests: page.items,
state: command.state,
output: command.output,
})
.map_err(PullRequestCommandError::Service)
})
.await
.map_err(|_| PullRequestCommandError::Unavailable)?
}
async fn run_pull_request_create(
repositories: Option<Arc<GitRepositories>>,
command: PullRequestCreateChannel,
) -> Result<PullRequestMutationResult, PullRequestCommandError> {
let (title, body) =
parse_issue_input(&command.input).map_err(|_| PullRequestCommandError::Input)?;
let repositories = repositories.ok_or(PullRequestCommandError::Unavailable)?;
let database = repositories
.push_database()
.ok_or(PullRequestCommandError::Unavailable)?
.to_owned();
let root = repositories.repository_root().to_owned();
let permit = repositories
.blocking_permit()
.await
.map_err(|_| PullRequestCommandError::Unavailable)?;
tokio::task::spawn_blocking(move || {
let _permit = permit;
PullRequestService::new(&database, &root)
.open(
&command.owner,
&command.repository,
&command.actor,
&title,
&body,
&command.base,
&command.head,
)
.map(|pull_request| PullRequestMutationResult {
owner: command.owner,
repository: command.repository,
number: pull_request.number,
operation: "created",
pull_request: Some(pull_request),
output: command.output,
})
.map_err(PullRequestCommandError::Service)
})
.await
.map_err(|_| PullRequestCommandError::Unavailable)?
}
async fn run_pull_request_set_state(
repositories: Option<Arc<GitRepositories>>,
actor: String,
command: PullRequestStateCommand,
) -> Result<PullRequestMutationResult, PullRequestCommandError> {
let repositories = repositories.ok_or(PullRequestCommandError::Unavailable)?;
let database = repositories
.push_database()
.ok_or(PullRequestCommandError::Unavailable)?
.to_owned();
let root = repositories.repository_root().to_owned();
let permit = repositories
.blocking_permit()
.await
.map_err(|_| PullRequestCommandError::Unavailable)?;
tokio::task::spawn_blocking(move || {
let _permit = permit;
PullRequestService::new(&database, &root)
.set_state(
&command.owner,
&command.repository,
command.number,
&actor,
command.state,
)
.map(|()| PullRequestMutationResult {
owner: command.owner,
repository: command.repository,
number: command.number,
operation: if command.state == "closed" {
"closed"
} else {
"reopened"
},
pull_request: None,
output: command.output,
})
.map_err(PullRequestCommandError::Service)
})
.await
.map_err(|_| PullRequestCommandError::Unavailable)?
}
fn send_pull_request_checkout_result(
channel: ChannelId,
result: Result<PullRequestCheckoutResult, PullRequestCommandError>,
machine_requested: bool,
session: &mut Session,
) -> Result<(), russh::Error> {
match result {
Ok(result) => {
let fetch = format!(
"git fetch origin refs/pull/{}/head:refs/heads/pr-{}",
result.number, result.number
);
let checkout = format!("git checkout pr-{}", result.number);
let data = match result.output {
CommandOutput::Human => format!("{fetch}\n{checkout}\n"),
CommandOutput::Json => format!(
"{{\"version\":1,\"status\":\"success\",\"repository\":{{\"owner\":\"{}\",\"name\":\"{}\"}},\"pull_request\":{{\"number\":{},\"ref\":\"refs/pull/{}/head\",\"local_branch\":\"pr-{}\"}},\"commands\":{{\"fetch\":\"{}\",\"checkout\":\"{}\"}}}}\n",
result.owner,
result.repository,
result.number,
result.number,
result.number,
fetch,
checkout,
),
};
session.data(channel, data.into_bytes())?;
finish_git_channel(channel, 0, session)
}
Err(error) => send_pull_request_error(channel, error, machine_requested, session),
}
}
fn send_pull_request_list_result(
channel: ChannelId,
result: Result<PullRequestListResult, PullRequestCommandError>,
machine_requested: bool,
session: &mut Session,
) -> Result<(), russh::Error> {
match result {
Ok(result) => {
let data = match result.output {
CommandOutput::Human => {
if result.pull_requests.is_empty() {
b"No pull requests.\n".to_vec()
} else {
result
.pull_requests
.iter()
.map(|pull_request| {
format!(
"#{} {} {} ({} <- {})\n",
pull_request.number,
pull_request.state,
pull_request.title,
pull_request.base_ref,
pull_request.head_ref
)
})
.collect::<String>()
.into_bytes()
}
}
CommandOutput::Json => {
let pull_requests = result
.pull_requests
.iter()
.map(pull_request_json)
.collect::<Vec<_>>();
json_line(serde_json::json!({
"version": 1,
"status": "success",
"repository": {
"owner": result.repository.owner,
"name": result.repository.slug,
},
"state": result.state,
"pull_requests": pull_requests,
}))
}
};
session.data(channel, data)?;
finish_git_channel(channel, 0, session)
}
Err(error) => send_pull_request_error(channel, error, machine_requested, session),
}
}
fn send_pull_request_mutation_result(
channel: ChannelId,
result: Result<PullRequestMutationResult, PullRequestCommandError>,
machine_requested: bool,
session: &mut Session,
) -> Result<(), russh::Error> {
match result {
Ok(result) => {
let data = match result.output {
CommandOutput::Human => format!(
"{} pull request {}/{}#{}.\n",
sentence_case(result.operation),
result.owner,
result.repository,
result.number
)
.into_bytes(),
CommandOutput::Json => json_line(serde_json::json!({
"version": 1,
"status": "success",
"repository": {
"owner": result.owner,
"name": result.repository,
},
"pull_request": result.pull_request.as_ref().map(pull_request_json)
.unwrap_or_else(|| serde_json::json!({ "number": result.number })),
"operation": result.operation,
})),
};
session.data(channel, data)?;
finish_git_channel(channel, 0, session)
}
Err(error) => send_pull_request_error(channel, error, machine_requested, session),
}
}
fn pull_request_json(pull_request: &crate::store::PullRequestRecord) -> serde_json::Value {
serde_json::json!({
"number": pull_request.number,
"title": pull_request.title,
"state": pull_request.state,
"author": pull_request.author,
"base": pull_request.base_ref,
"head": pull_request.head_ref,
"created_at": pull_request.created_at,
"updated_at": pull_request.updated_at,
})
}
fn send_pull_request_error(
channel: ChannelId,
error: PullRequestCommandError,
machine_requested: bool,
session: &mut Session,
) -> Result<(), russh::Error> {
if machine_requested {
session.data(
channel,
json_line(serde_json::json!({
"version": 1,
"status": "error",
"error": { "code": pull_request_command_error_code(&error) },
})),
)?;
} else {
session.extended_data(
channel,
1,
format!("tit: {}\n", pull_request_command_error_message(&error)).into_bytes(),
)?;
}
finish_git_channel(channel, 1, session)
}
fn pull_request_command_error_code(error: &PullRequestCommandError) -> &'static str {
match error {
PullRequestCommandError::Usage => "invalid-command",
PullRequestCommandError::Input
| PullRequestCommandError::Service(
PullRequestError::Title | PullRequestError::Body | PullRequestError::Branch,
) => "invalid-input",
PullRequestCommandError::Unavailable => "service-unavailable",
PullRequestCommandError::Store(
StoreError::PullRequestHidden
| StoreError::PullRequestNotFound(_, _, _)
| StoreError::RepositoryNotFound(_, _),
) => "pull-request-unavailable",
PullRequestCommandError::Service(PullRequestError::Auth(_))
| PullRequestCommandError::Service(PullRequestError::RepositoryName(_)) => "invalid-target",
PullRequestCommandError::Service(PullRequestError::Store(
StoreError::PullRequestHidden
| StoreError::PullRequestNotFound(_, _, _)
| StoreError::RepositoryNotFound(_, _),
)) => "pull-request-unavailable",
PullRequestCommandError::Service(PullRequestError::Store(
StoreError::PullRequestDenied,
)) => "permission-denied",
PullRequestCommandError::Store(_) | PullRequestCommandError::Service(_) => {
"pull-request-command-failed"
}
}
}
fn pull_request_command_error_message(error: &PullRequestCommandError) -> String {
match pull_request_command_error_code(error) {
"invalid-command" => {
format!(
"usage: {PULL_REQUEST_LIST_USAGE}, {PULL_REQUEST_CREATE_USAGE}, \
{PULL_REQUEST_CLOSE_USAGE}, {PULL_REQUEST_REOPEN_USAGE}, or \
{PULL_REQUEST_CHECKOUT_USAGE}\n{HELP_GUIDANCE}"
)
}
"invalid-input" => {
"The refs or input are not valid. The first input line is the title and the remaining input is the body."
.to_owned()
}
"pull-request-unavailable" => "The pull request is not available.".to_owned(),
"invalid-target" => "The pull-request target is not valid.".to_owned(),
"permission-denied" => {
"The account cannot change a pull request in this repository.".to_owned()
}
"service-unavailable" => "The pull-request service is not available.".to_owned(),
_ => "The pull-request command could not be completed.".to_owned(),
}
}
impl SshSession {
fn authorize(&self, public_key: &PublicKey) -> Auth {
if self.authorized_keys.contains(public_key) {
Auth::Accept
} else {
Auth::reject()
}
}
async fn open_git_service(&mut self, command: &[u8]) -> Option<InitialGitService> {
let active_channels = self
.exec_channels
.values()
.filter(|channel| matches!(channel, ExecChannel::Upload(_) | ExecChannel::Receive(_)))
.count();
let global_permit = reserve_git_channel(active_channels, &self.git_channels)?;
let repositories = self.repositories.as_ref()?;
let identity = self.active_identity()?;
let service = repositories
.resolve_ssh_service_for(Some(&identity.username), command)
.ok()?;
match service {
GitSshService::Upload { path, .. } => {
let permit = repositories.blocking_permit().await.ok()?;
let protocol = self.protocol;
tokio::task::spawn_blocking(move || {
let _permit = permit;
let service = UploadPack::open(&path)?;
let advertisement = service.advertisement(protocol, false)?;
Ok::<_, UploadPackError>(InitialGitService::Upload {
service: Box::new(service),
advertisement,
global_permit,
})
})
.await
.ok()?
.ok()
}
GitSshService::Receive {
path,
owner,
repository,
} => {
if !repositories.uses_policy() && !self.authenticated_writer {
return None;
}
let database = repositories.push_database()?.to_owned();
let actor = identity.username.clone();
let public_key = self.authenticated_key.clone()?;
let uses_policy = repositories.uses_policy();
let permit = repositories.blocking_permit().await.ok()?;
tokio::task::spawn_blocking(move || {
let _permit = permit;
let service = if uses_policy {
ReceivePack::open_authorized(
&path,
&database,
actor,
owner.clone(),
repository.clone(),
)?
} else {
ReceivePack::open(&path, &database, actor)?
};
let advertisement = service.advertisement()?;
Ok::<_, ReceivePackError>(InitialGitService::Receive(Box::new(
InitialReceiveService {
service: Box::new(service),
advertisement,
owner,
repository,
identity,
public_key,
global_permit,
},
)))
})
.await
.ok()?
.ok()
}
}
}
fn active_identity(&self) -> Option<SshIdentity> {
let public_key = self.authenticated_key.as_ref()?;
let authenticated = self.authenticated_identity.as_ref()?;
let current = self.authorized_keys.identity(public_key)?;
(current == *authenticated).then_some(current)
}
}
enum InitialGitService {
Upload {
service: Box<UploadPack>,
advertisement: Vec<u8>,
global_permit: OwnedSemaphorePermit,
},
Receive(Box<InitialReceiveService>),
}
struct InitialReceiveService {
service: Box<ReceivePack>,
advertisement: Vec<u8>,
owner: String,
repository: String,
identity: SshIdentity,
public_key: PublicKey,
global_permit: OwnedSemaphorePermit,
}
async fn receive_data(git: &mut ReceiveChannel, data: &[u8]) -> Result<(), ()> {
if git.commands_complete {
write_receive_pack(git, data).await?;
return Ok(());
}
git.commands.extend_from_slice(data);
let boundary = first_flush_end(&git.commands).map_err(|_| ())?;
let Some(boundary) = boundary else {
return Ok(());
};
let pack = git.commands.split_off(boundary);
git.commands_complete = true;
write_receive_pack(git, &pack).await
}
async fn write_receive_pack(git: &mut ReceiveChannel, data: &[u8]) -> Result<(), ()> {
let bytes = u64::try_from(data.len()).map_err(|_| ())?;
git.pack_bytes = git.pack_bytes.checked_add(bytes).ok_or(())?;
if git.pack_bytes > MAX_RECEIVE_PACK_BYTES {
return Err(());
}
git.pack.write_all(data).await.map_err(|_| ())
}
type ReceiveResult = Option<Result<Vec<u8>, (ReceivePackError, Vec<u8>)>>;
async fn finish_receive(
repositories: Option<Arc<GitRepositories>>,
git: Box<ReceiveChannel>,
) -> ReceiveResult {
let ReceiveChannel {
mut service,
owner,
repository,
identity,
public_key,
authorized_keys,
commands,
mut pack,
started_at,
..
} = *git;
if started_at.elapsed() > GIT_CHANNEL_WALL_CLOCK_LIMIT - Duration::from_secs(30) {
return None;
}
pack.flush().await.ok()?;
pack.sync_all().await.ok()?;
drop(pack);
let repositories = repositories?;
let push_permit = repositories.push_permit().await.ok()?;
let permit = repositories.blocking_permit().await.ok()?;
let maintenance = repositories.mutation_permit().await;
tokio::task::spawn_blocking(move || {
let _permit = permit;
let _push_permit = push_permit;
let _maintenance = maintenance;
if authorized_keys.identity(&public_key).as_ref() != Some(&identity)
|| !repositories.authorize(
&identity.username,
&owner,
&repository,
RepositoryOperation::Write,
)
{
return None;
}
Some(match service.finish(&commands) {
Ok(response) => Ok(response),
Err(error) => {
let response = service.rejection_response(&commands, &error);
match service.record_rejection() {
Ok(()) => Err((error, response)),
Err(audit_error) => {
let response = service.rejection_response(&commands, &audit_error);
Err((audit_error, response))
}
}
}
})
})
.await
.ok()
.flatten()
}
fn send_receive_result(
channel: ChannelId,
result: ReceiveResult,
session: &mut Session,
) -> Result<(), russh::Error> {
match result {
Some(Ok(response)) => {
session.data(channel, response)?;
finish_git_channel(channel, 0, session)
}
Some(Err((error, response))) => {
eprintln!("tit: receive-pack failed: {error}");
session.data(
channel,
if response.is_empty() {
receive_error_response()
} else {
response
},
)?;
finish_git_channel(channel, 1, session)
}
None => {
session.data(channel, receive_error_response())?;
finish_git_channel(channel, 1, session)
}
}
}
fn receive_error_response() -> Vec<u8> {
let mut response = Vec::new();
let _ = encode_data(b"unpack tit rejected the push\n", &mut response);
super::git::packetline::encode_flush(&mut response);
response
}
async fn respond_git(
repositories: Option<Arc<GitRepositories>>,
protocol: ProtocolVersion,
git: Box<UploadChannel>,
) -> Option<(Box<UploadChannel>, Result<Vec<u8>, UploadPackError>)> {
let remaining = GIT_CHANNEL_WALL_CLOCK_LIMIT.checked_sub(git.started_at.elapsed())?;
let cancelled = Arc::clone(&git.cancelled);
let permit = repositories?.blocking_permit().await.ok()?;
let cancellation_for_job = Arc::clone(&cancelled);
let mut task = tokio::task::spawn_blocking(move || {
let _permit = permit;
let response =
git.service
.respond_with_cancellation(protocol, &git.request, &cancellation_for_job);
(git, response)
});
match tokio::time::timeout(remaining, &mut task).await {
Ok(result) => result.ok(),
Err(_) => {
cancelled.store(true, Ordering::Relaxed);
task.await.ok()
}
}
}
fn git_response_within_limit(git: &UploadChannel, response_bytes: usize) -> bool {
let response_bytes = u64::try_from(response_bytes).unwrap_or(u64::MAX);
git_channel_bytes_within_limit(git.transferred, response_bytes)
}
fn git_channel_bytes_within_limit(transferred: u64, additional: u64) -> bool {
transferred.saturating_add(additional) <= MAX_GIT_CHANNEL_BYTES
}
fn valid_git_protocol(value: &str) -> bool {
matches!(value, "version=0" | "version=1" | "version=2")
}
fn trim_line(line: &[u8]) -> &[u8] {
line.strip_suffix(b"\n").unwrap_or(line)
}
fn finish_git_channel(
channel: ChannelId,
status: u32,
session: &mut Session,
) -> Result<(), russh::Error> {
session.exit_status_request(channel, status)?;
session.eof(channel)?;
session.close(channel)?;
Ok(())
}
fn fail_git_channel(channel: ChannelId, session: &mut Session) -> Result<(), russh::Error> {
let mut error = Vec::new();
encode_data(b"ERR invalid Git request\n", &mut error)
.expect("a Git error packet is within the limit");
session.data(channel, error)?;
finish_git_channel(channel, 1, session)
}
#[derive(Default)]
struct RequestAudit {
accepted_env: AtomicUsize,
rejected_env: AtomicUsize,
accepted_exec: AtomicUsize,
rejected_exec: AtomicUsize,
rejected_shell: AtomicUsize,
rejected_pty: AtomicUsize,
rejected_agent: AtomicUsize,
rejected_forward: AtomicUsize,
}
impl RequestAudit {
fn snapshot(&self) -> RequestAuditSnapshot {
RequestAuditSnapshot {
accepted_env: self.accepted_env.load(Ordering::Relaxed),
rejected_env: self.rejected_env.load(Ordering::Relaxed),
accepted_exec: self.accepted_exec.load(Ordering::Relaxed),
rejected_exec: self.rejected_exec.load(Ordering::Relaxed),
rejected_shell: self.rejected_shell.load(Ordering::Relaxed),
rejected_pty: self.rejected_pty.load(Ordering::Relaxed),
rejected_agent: self.rejected_agent.load(Ordering::Relaxed),
rejected_forward: self.rejected_forward.load(Ordering::Relaxed),
}
}
}
#[derive(Clone, Copy, Debug, Default)]
pub(crate) struct RequestAuditSnapshot {
pub(crate) accepted_env: usize,
pub(crate) rejected_env: usize,
pub(crate) accepted_exec: usize,
pub(crate) rejected_exec: usize,
pub(crate) rejected_shell: usize,
pub(crate) rejected_pty: usize,
pub(crate) rejected_agent: usize,
pub(crate) rejected_forward: usize,
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn parses_only_bounded_issue_commands_and_input() {
assert!(matches!(
parse_issue_command(b"issue list alice/project --output json"),
Ok(IssueCommand::List(_))
));
assert!(matches!(
parse_issue_command(b"issue create alice/project"),
Ok(IssueCommand::Create(_))
));
for command in [
b"issue list alice/project --output json extra".as_slice(),
b"issue list alice/project/extra".as_slice(),
b"issue create alice/project --output json --output json".as_slice(),
&[b'x'; MAX_ISSUE_COMMAND_BYTES + 1],
] {
assert!(parse_issue_command(command).is_err());
}
let mut input = Vec::new();
assert!(append_issue_input(&mut input, &[b'x'; MAX_ISSUE_INPUT_BYTES]).is_ok());
assert!(append_issue_input(&mut input, b"x").is_err());
assert!(matches!(
parse_issue_input(b"invalid\xff"),
Err(IssueCommandError::Input)
));
}
#[test]
fn parses_only_bounded_pull_request_commands() {
let command = parse_pull_request_command(b"pr checkout alice/project 42 --output json")
.expect("parse a pull-request checkout command");
let PullRequestCommand::Checkout(command) = command else {
panic!("expected checkout");
};
assert_eq!(command.owner, "alice");
assert_eq!(command.repository, "project");
assert_eq!(command.number, 42);
assert!(command.output == CommandOutput::Json);
assert!(matches!(
parse_pull_request_command(
b"pr list alice/project --output json --state merged"
),
Ok(PullRequestCommand::List(PullRequestListCommand {
state,
output: CommandOutput::Json,
..
})) if state == "merged"
));
assert!(matches!(
parse_pull_request_command(b"pr create alice/project main topic"),
Ok(PullRequestCommand::Create(_))
));
assert!(matches!(
parse_pull_request_command(b"pr close alice/project 2"),
Ok(PullRequestCommand::SetState(_))
));
for command in [
b"pr checkout alice/project 0".as_slice(),
b"pr checkout alice/project not-a-number".as_slice(),
b"pr checkout alice/project/extra 1".as_slice(),
b"pr checkout alice/project 1 --output json extra".as_slice(),
b"pr checkout alice/project 1 --output json --output json".as_slice(),
b"pr list alice/project --state invalid".as_slice(),
b"pr create alice/project main".as_slice(),
&[b'x'; MAX_PULL_REQUEST_COMMAND_BYTES + 1],
] {
assert!(parse_pull_request_command(command).is_err());
}
}
#[test]
fn bounds_git_channel_time_bytes_and_concurrency() {
let now = Instant::now();
assert!(!git_channel_expired(now, now));
assert!(git_channel_expired(
now - GIT_CHANNEL_WALL_CLOCK_LIMIT - Duration::from_secs(1),
now
));
assert!(git_channel_expired(
now,
now - GIT_CHANNEL_INACTIVITY_LIMIT - Duration::from_secs(1)
));
let global = Arc::new(Semaphore::new(MAX_GIT_CHANNELS_GLOBAL));
let permits = (0..MAX_GIT_CHANNELS_GLOBAL)
.map(|_| reserve_git_channel(0, &global).expect("reserve a global Git channel"))
.collect::<Vec<_>>();
assert!(reserve_git_channel(0, &global).is_none());
drop(permits);
assert!(reserve_git_channel(MAX_GIT_CHANNELS_PER_CONNECTION, &global).is_none());
let limit = std::hint::black_box(MAX_GIT_CHANNEL_BYTES);
assert!(git_channel_bytes_within_limit(limit, 0));
assert!(!git_channel_bytes_within_limit(limit, 1));
}
}