use std::collections::BTreeSet;
use std::process::Command;
use std::sync::Arc;
use anyhow::Context as _;
use knot_acl::{KnotAcl, can_push};
use knot_git::RefClass;
use knot_index::Resolved;
use knot_types::{AdmissionPolicy, ClonePath, OwnerDid, OwnerRef, RepoDid};
use crate::config::ValidatedConfig;
use crate::manifest;
use crate::out::{self, out};
struct GitCommand {
verb: Verb,
path: RepoPath,
}
enum RepoPath {
Did(RepoDid),
Owner(OwnerRef, String),
Rkey(String),
}
impl std::fmt::Display for RepoPath {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::Did(did) => write!(f, "{did}"),
Self::Owner(_, name) | Self::Rkey(name) => write!(f, "{name}"),
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Verb {
UploadPack,
ReceivePack,
}
impl Verb {
fn is_write(self) -> bool {
self == Verb::ReceivePack
}
}
/// only upload-pack and receive-pack pass, a forced command must never exec client input.
fn parse_command(raw: &str) -> anyhow::Result<GitCommand> {
anyhow::ensure!(
!raw.contains(char::is_control),
"SSH_ORIGINAL_COMMAND can't hold control characters"
);
let (head, rest) = raw
.split_once(' ')
.context("no git command in SSH_ORIGINAL_COMMAND (shell access is not provided)")?;
let (verb_str, quoted_raw) = match head {
"git" => rest
.split_once(' ')
.context("incomplete git command in SSH_ORIGINAL_COMMAND")?,
verb => (verb.strip_prefix("git-").unwrap_or(verb), rest),
};
let verb = match verb_str {
"upload-pack" => Verb::UploadPack,
"receive-pack" => Verb::ReceivePack,
other => anyhow::bail!("unsupported command {other:?}: only git fetch/push is served"),
};
let quoted = quoted_raw.trim();
let repo_path = quoted
.strip_prefix('\'')
.and_then(|s| s.strip_suffix('\''))
.unwrap_or(quoted);
let key = repo_path.trim_start_matches('/').trim_end_matches('/');
let unsuffixed = key.strip_suffix(".git").unwrap_or(key);
let path = match key.split_once('/') {
Some((owner, name)) => RepoPath::Owner(
OwnerRef::parse(owner)
.context("the owner in the repo path is neither a DID nor a handle")?,
name.to_string(),
),
None if unsuffixed.starts_with("did:") => RepoPath::Did(RepoDid::new(unsuffixed)?),
None if unsuffixed.is_empty() => anyhow::bail!("empty repo path"),
None => RepoPath::Rkey(unsuffixed.to_string()),
};
Ok(GitCommand { verb, path })
}
fn ref_snapshot(repo_path: &std::path::Path) -> anyhow::Result<String> {
let out = Command::new("git")
.arg("-C")
.arg(repo_path)
.args(["for-each-ref", "--format=%(objectname) %(refname)"])
.output()
.context("snapshot repository refs")?;
anyhow::ensure!(
out.status.success(),
"ref snapshot failed: {}",
String::from_utf8_lossy(&out.stderr)
);
Ok(String::from_utf8_lossy(&out.stdout).into_owned())
}
const PUSH_DENIED: &str = "knot: you aren't authorized to push to this repository.";
pub enum Pusher {
Operator,
Accounts(Vec<knot_types::AccountDid>),
}
impl Pusher {
/// only `--operator` makes the operator, and dropped args fail here instead of promoting a key.
pub fn from_flags(operator: bool, actors: Vec<String>) -> anyhow::Result<Self> {
anyhow::ensure!(
operator || !actors.is_empty(),
"git-ssh needs --operator or --actor DID"
);
if operator {
return Ok(Self::Operator);
}
Ok(Self::Accounts(
actors
.into_iter()
.map(knot_types::AccountDid::new)
.collect::<Result<_, _>>()?,
))
}
fn pushes_as(
&self,
config: &ValidatedConfig,
index: &knot_index::Index,
repo: &RepoDid,
owner: &knot_types::AccountDid,
) -> anyhow::Result<Option<knot_types::AccountDid>> {
match self {
Self::Operator => Ok(Some(owner.clone())),
Self::Accounts(actors) => {
for actor in actors {
if may_push(config, index, std::slice::from_ref(actor), repo)? {
return Ok(Some(actor.clone()));
}
}
Ok(None)
}
}
}
fn authorizes_as(&self, config: &ValidatedConfig, owner: &knot_types::AccountDid) -> bool {
match self {
Self::Operator => owner == &config.owner,
Self::Accounts(actors) => actors.contains(owner),
}
}
fn provisions(&self, config: &ValidatedConfig) -> bool {
match self {
Self::Operator => true,
Self::Accounts(actors) => actors.contains(&config.owner),
}
}
}
async fn owner_did(config: &ValidatedConfig, owner: &OwnerRef) -> anyhow::Result<OwnerDid> {
match owner {
OwnerRef::Did(did) => Ok(did.clone()),
OwnerRef::Handle(handle) => {
if let Some(did) = crate::datastore::Datastore::at(&config.output_root)
.record::<crate::datastore::schema::OwnerHandles>(&())
.read()
.ok()
.flatten()
.and_then(|stored| {
stored
.entries
.into_iter()
.find(|(_, entry)| entry.handle.eq_ignore_ascii_case(handle.as_str()))
.and_then(|(did, _)| OwnerDid::new(did).ok())
})
{
return Ok(did);
}
let transport = knot_runtime::ReqwestHttp::new(config.http_limits())?;
let did = config
.atproto(transport)?
.resolve_handle_to_did(handle)
.await
.with_context(|| format!("resolve the handle {handle}"))?;
Ok(did.into())
}
}
}
async fn hosted_repo_for_path(
index: &knot_index::Index,
path: &RepoPath,
owner: Option<&OwnerDid>,
) -> anyhow::Result<Option<RepoDid>> {
match (path, owner) {
(RepoPath::Did(did), _) => {
Ok(matches!(index.ownership_of(did), Resolved::Ready(Some(_))).then(|| did.clone()))
}
(RepoPath::Owner(_, name), Some(owner)) => {
let clone = ClonePath::parse(name).context("invalid repo name")?;
match index.resolve_clone_path(owner, &clone) {
Resolved::Ready(found) => Ok(found),
Resolved::Warming => anyhow::bail!("the registry is still warming, try again"),
}
}
(RepoPath::Owner(..), None) => unreachable!("an owner path resolves its owner first"),
(RepoPath::Rkey(rkey), _) => {
let mut named = crate::registry::matching_bare_name(index, rkey)?.into_iter();
let first = named.next();
anyhow::ensure!(
named.next().is_none(),
"{rkey} names more than one repo here, use <owner>/{rkey} or the repo's DID"
);
Ok(first)
}
}
}
async fn resolve_or_provision(
config: &Arc<ValidatedConfig>,
git_cmd: &GitCommand,
pusher: &Pusher,
) -> anyhow::Result<(RepoDid, knot_types::AccountDid, knot_types::AccountDid)> {
let layout = config.layout()?;
let meta_path = layout.meta_path(&config.knot_did)?;
let index = knot_index::Index::new(&meta_path, layout);
index.rebuild().context("rebuild registry index")?;
let owner = match &git_cmd.path {
RepoPath::Owner(owner, _) => Some(owner_did(config, owner).await?),
_ => None,
};
if let Some(did) = hosted_repo_for_path(&index, &git_cmd.path, owner.as_ref()).await?
&& let Resolved::Ready(Some(ownership)) = index.ownership_of(&did)
{
let owner_account = knot_types::AccountDid::new(ownership.owner.to_string())
.context("invalid owner DID in index")?;
let actor = if git_cmd.verb.is_write() {
pusher
.pushes_as(config, &index, &did, &owner_account)?
.context(PUSH_DENIED)?
} else {
owner_account.clone()
};
return Ok((did, owner_account, actor));
}
anyhow::ensure!(
git_cmd.verb.is_write(),
"repo {} is not hosted on this knot (fetch of unknown repo)",
git_cmd.path
);
let rkey = match (&git_cmd.path, &owner) {
(RepoPath::Did(did), _) => anyhow::bail!("repo {did} is not hosted on this knot"),
(RepoPath::Owner(_, name), Some(owner)) => {
anyhow::ensure!(
owner.is(&config.owner),
"repo {} is not hosted on this knot",
git_cmd.path
);
name.strip_suffix(".git").unwrap_or(name).to_string()
}
(RepoPath::Owner(..), None) => unreachable!("an owner path resolves its owner first"),
(RepoPath::Rkey(rkey), _) => rkey.clone(),
};
anyhow::ensure!(pusher.provisions(config), PUSH_DENIED);
out!("repo {rkey} is new: provisioning (minting repo DID, submitting genesis)...");
let did = crate::provision::create_for_push(config.clone(), rkey, None).await?;
Ok((did, config.owner.clone(), config.owner.clone()))
}
pub async fn run(config: Arc<ValidatedConfig>, pusher: Pusher) -> anyhow::Result<()> {
out::as_transport();
let raw = std::env::var("SSH_ORIGINAL_COMMAND").unwrap_or_default();
let served = transport(config, pusher, &raw).await;
// a fetch client only reads pkt-lines, and the refusal goes out as an `ERR` line.
if let Err(error) = &served
&& !raw.contains("receive-pack")
{
let _ = crate::ssh_fetch::refuse(&mut std::io::stdout().lock(), &format!("{error:#}"));
}
served
}
async fn transport(config: Arc<ValidatedConfig>, pusher: Pusher, raw: &str) -> anyhow::Result<()> {
let git_cmd = parse_command(raw)?;
let (repo_did, repo_owner, actor) = resolve_or_provision(&config, &git_cmd, &pusher).await?;
let layout = config.layout()?;
let repo_path = layout
.guarded_path(&repo_did)
.with_context(|| format!("resolve guarded path for {repo_did}"))?;
anyhow::ensure!(repo_path.exists(), "hosted repo {repo_did} missing on disk");
if git_cmd.verb.is_write() {
let asks = pusher.authorizes_as(&config, &repo_owner);
let oauth = crate::oauth::ensure_auth_for_owner(&config, &repo_owner, asks).await;
let (status, moved) = receive_locked(&config, &repo_did, &repo_path, || {
receive_pack(&config)
.arg(&repo_path)
.status()
.context("spawn git transport process")
})?;
if !moved.is_empty() {
match oauth {
Err(e) => {
out!("oauth pre-hook notice: {e:#}");
out!("the repo declaration cannot be published automatically");
}
Ok(None) => out!(
"the repo declaration isn't published: {repo_owner} has no oauth session here, so their next push or `knot-static oauth login --owner {repo_owner}` publishes it"
),
Ok(Some(_)) => {}
}
out!("push accepted, building...");
match crate::writes::publish(&config, &repo_did).await {
Ok(changed) => {
let heard =
crate::spindle::tell(&config, &changed, &repo_did, &actor, &moved).await;
for line in &heard.said {
out!("{line}");
}
notify_post_activate(&config, &changed, heard.pending, true);
}
Err(e) => {
out!("warning: the push wasn't published: {e:#}");
crate::spindle::defer(&config, &repo_did, &actor, &moved).await;
}
}
}
anyhow::ensure!(status.success(), "git receive-pack exited with {status}");
return Ok(());
}
crate::ssh_fetch::serve(
&config,
&repo_did,
std::env::var("GIT_PROTOCOL").ok().as_deref(),
crate::ssh_fetch::Deadline::new(
std::io::stdin().lock(),
crate::ssh_fetch::REQUEST_DEADLINE,
),
std::io::stdout().lock(),
)
}
pub(crate) fn may_push(
config: &ValidatedConfig,
index: &knot_index::Index,
actors: &[knot_types::AccountDid],
repo: &RepoDid,
) -> Result<bool, knot_index::IndexError> {
index.ensure_collaborators(repo)?;
let admins = BTreeSet::from([config.owner.clone()]);
let acl = KnotAcl::new(&admins, AdmissionPolicy::Closed, index);
Ok(actors
.iter()
.any(|actor| can_push(&acl, &index.collaboration_consent(repo, actor)).is_allowed()))
}
pub(crate) const REPLACE_REFS: &str = "refs/replace";
/// non-public refs (cobs, checkpoints) are hidden, and cobs only move through the cob write path.
/// replace refs stay hidden, since git would follow a pushed one and clients wouldn't.
/// git and whatever it spawns while it reads a pusher's pack never see the master key.
pub(crate) fn receive_pack(config: &ValidatedConfig) -> Command {
let mut git = Command::new("git");
crate::git_export::true_history(&mut git)
.env_remove(&config.master_key_env)
.env_remove("KNOT_MASTER_KEY")
.env("GIT_CONFIG_GLOBAL", "/dev/null")
.env("GIT_CONFIG_SYSTEM", "/dev/null")
.env("GIT_CONFIG_NOSYSTEM", "1")
.env("LC_ALL", "C");
for setting in [
"core.hooksPath=/dev/null",
"receive.autogc=false",
"gc.auto=0",
"maintenance.auto=false",
"receive.fsckObjects=true",
] {
git.args(["-c", setting]);
}
for namespace in RefClass::ALL
.into_iter()
.filter(|class| !class.is_public())
.filter_map(RefClass::namespace)
{
git.args(["-c", &format!("receive.hideRefs={namespace}")]);
}
git.args(["-c", &format!("receive.hideRefs={REPLACE_REFS}")]);
git.arg("receive-pack");
git
}
pub(crate) fn receive_locked(
config: &ValidatedConfig,
repo_did: &RepoDid,
repo_path: &std::path::Path,
transport: impl FnOnce() -> anyhow::Result<std::process::ExitStatus>,
) -> anyhow::Result<(std::process::ExitStatus, Vec<crate::spindle::RefMove>)> {
// exclusive, because a build capture must never see refs move under it.
let repo_key = crate::plan::repo_key(repo_did.as_str());
let lock = if let Some(lock) = crate::repolock::try_acquire(
&config.output_root,
&repo_key,
crate::repolock::LockMode::Exclusive,
)? {
lock
} else {
out!("waiting for the repo lock ({repo_did})");
crate::repolock::acquire(
&config.output_root,
&repo_key,
crate::repolock::LockMode::Exclusive,
)?
};
tracing::debug!(
target: "knot_static::git_ssh",
repo = lock.key(),
mode = lock.mode().describe(),
path = %lock.path().display(),
"holding the repository source lock for the transport"
);
// a no-op push still exits 0, and it should not start a generation.
let refs_before = ref_snapshot(repo_path)?;
let status = transport()?;
if status.success() {
crate::commit_graph::refresh(repo_path);
}
// flock isn't recursive and publishing takes the same lock, release it first.
drop(lock);
let moved = if status.success() {
crate::spindle::moves(&refs_before, &ref_snapshot(repo_path)?)
} else {
Vec::new()
};
Ok((status, moved))
}
pub(crate) fn notify_post_activate(
config: &ValidatedConfig,
changed_repos: &[String],
drain: bool,
declarations: bool,
) {
if (!changed_repos.is_empty() || drain)
&& let Err(e) = spawn_detached(config, post_activate_args(changed_repos, declarations))
{
out!("warning: failed to detach post-activation tasks: {e:#}");
}
}
pub(crate) fn post_activate_args(changed_repos: &[String], declarations: bool) -> Vec<&str> {
let skip = (!declarations).then_some("--no-declarations");
let repos = changed_repos
.iter()
.flat_map(|repo| ["--repo", repo.as_str()]);
["post-activate"]
.into_iter()
.chain(skip)
.chain(repos)
.collect()
}
pub(crate) async fn activate_with_merge(
config: Arc<ValidatedConfig>,
generation_id: &str,
repo_hint: Option<&str>,
) -> anyhow::Result<Vec<String>> {
crate::failpoint::pause("before_activation");
let first = match crate::failpoint::check("merge_rebuild") {
Ok(()) => crate::activate::activate(config.clone(), generation_id).await,
Err(_) => Err(crate::publish::StalePlan::injected().into()),
};
match first {
Ok(result) => {
out!(
"generation {} activated",
manifest::short_id(&result.generation)
);
Ok(result.changed_repos)
}
Err(err) if err.downcast_ref::<crate::publish::StalePlan>().is_some() => {
out!("candidate went stale (another push activated first), rebuilding...");
let rebuilt = match crate::failpoint::check("merge_rebuild") {
Ok(()) => crate::generate::run(config.clone(), repo_hint).await,
Err(error) => Err(error),
};
match rebuilt {
Ok(crate::generate::BuildOutcome::Built {
generation_id,
pin: _pin,
..
}) => match crate::activate::activate(config.clone(), &generation_id).await {
Ok(result) => {
out!(
"generation {} activated after merge",
manifest::short_id(&result.generation)
);
Ok(result.changed_repos)
}
Err(second) => Err(second.context("activation failed after the merge rebuild")),
},
Ok(crate::generate::BuildOutcome::Unchanged { generation_id }) => {
out!(
"generation {} already covers the merged state",
manifest::short_id(&generation_id)
);
Ok(Vec::new())
}
Err(build_err) => Err(build_err.context("the merge rebuild failed")),
}
}
Err(err) => Err(err),
}
}
pub(crate) fn open_log_sink(config: &ValidatedConfig, name: &str) -> std::fs::File {
use std::path::PathBuf;
let candidates = [
PathBuf::from("/var/log/knot-static").join(name),
PathBuf::from("/var/log/knot-static.log"),
config.output_root.join("logs").join(name),
std::env::temp_dir().join(format!("knot-static-{name}")),
];
for path in &candidates {
if let Some(parent) = path.parent() {
let _ = std::fs::create_dir_all(parent);
}
if let Ok(file) = std::fs::OpenOptions::new()
.create(true)
.append(true)
.open(path)
{
return file;
}
}
std::fs::OpenOptions::new()
.write(true)
.open("/dev/null")
.expect("open /dev/null fallback")
}
pub(crate) fn spawn_detached<'a>(
config: &ValidatedConfig,
args: impl IntoIterator<Item = &'a str>,
) -> anyhow::Result<()> {
let exe = std::env::current_exe().context("resolve current executable")?;
let use_systemd_cat = std::path::Path::new("/run/systemd/journal/socket").exists()
&& Command::new("systemd-cat")
.arg("--version")
.output()
.map(|o| o.status.success())
.unwrap_or(false);
let mut cmd = if use_systemd_cat {
let mut c = Command::new("systemd-cat");
c.arg("-t").arg("knot-static");
c.arg(&exe);
c
} else {
Command::new(&exe)
};
cmd.arg("--config").arg(&config.config_path);
cmd.args(args);
cmd.stdin(std::process::Stdio::null());
if use_systemd_cat {
cmd.stdout(std::process::Stdio::null());
cmd.stderr(std::process::Stdio::null());
} else {
let sink = open_log_sink(config, "post-activate.log");
let sink_out = sink.try_clone().context("clone log sink for stdout")?;
cmd.stdout(sink_out);
cmd.stderr(sink);
}
#[cfg(unix)]
{
use std::os::unix::process::CommandExt;
cmd.process_group(0);
}
cmd.spawn().context("spawn background process")?;
Ok(())
}
#[cfg(test)]
#[path = "tests.rs"]
mod tests;