a73x

21747153

Say who is speaking on every line a connection logs

a73x   2026-08-18 19:22

Commit message
Say who is speaking on every line a connection logs

Only the auth line named the principal. Everything after it — the command
being run, the reason it was refused — was anonymous, so with two
connections interleaved no line could be attributed to either of them.

A span carrying the principal, entered by every handler body, puts the
identity on all of them: the fingerprint for a key session, "<person> (via
<key-id>)" for a certificate. That form is now produced once, where the
certificate validates, instead of being spelled one way at auth and another
way per command.

The attribution test used to pass on the auth line alone. It now reads the
line naming the git command, which can carry an identity only if the command
itself is attributed.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
src/server/ssh/session.rs
Old New
@@ -7,7 +7,7 @@ use russh::server::{Auth, ChannelOpenHandle, Handler, Msg, Session};
7 use russh::{Channel, ChannelId}; 7 use russh::{Channel, ChannelId};
8 use tokio::process::Command; 8 use tokio::process::Command;
9 use tokio::sync::mpsc; 9 use tokio::sync::mpsc;
10 use tracing::{debug, error, info, warn}; 10 use tracing::{debug, error, info, info_span, warn, Instrument, Span};
11 11
12 use super::auth::{is_authorized, load_authorized_keys}; 12 use super::auth::{is_authorized, load_authorized_keys};
13 use crate::governance::conf::{Access, Subject}; 13 use crate::governance::conf::{Access, Subject};
@@ -52,7 +52,14 @@ enum AuthIdentity {
52 /// A delegate certificate. Kept whole so the regime can re-validate it 52 /// A delegate certificate. Kept whole so the regime can re-validate it
53 /// per request — expiry and cadir/keydir membership are checked on every 53 /// per request — expiry and cadir/keydir membership are checked on every
54 /// command, not once at connection open. 54 /// command, not once at connection open.
55 Delegate { certificate: Box<Certificate> }, 55 Delegate {
56 certificate: Box<Certificate>,
57 /// `<person> (via <key-id>)`, from the validation that accepted this
58 /// certificate. Re-validation decides what the connection may still
59 /// *do*; it cannot change who it authenticated as, so the one name a
60 /// log line should carry is settled here and never recomputed.
61 principal: String,
62 },
56 } 63 }
57 64
58 /// Per-connection SSH session handler. 65 /// Per-connection SSH session handler.
@@ -67,6 +74,11 @@ pub struct SshHandler {
67 stdin_tx: Option<(ChannelId, mpsc::Sender<Vec<u8>>)>, 74 stdin_tx: Option<(ChannelId, mpsc::Sender<Vec<u8>>)>,
68 /// In-progress release upload, fed by data() and finalized on channel EOF. 75 /// In-progress release upload, fed by data() and finalized on channel EOF.
69 active_upload: Option<UploadSession>, 76 active_upload: Option<UploadSession>,
77 /// Names this connection's principal on every line its handlers log.
78 /// Connections are concurrent and their lines interleave, so without this
79 /// only the one auth line could be attributed to anybody. `Span::none()`
80 /// until auth accepts — before that there is no one to name.
81 span: Span,
70 } 82 }
71 83
72 struct UploadSession { 84 struct UploadSession {
@@ -88,6 +100,7 @@ impl SshHandler {
88 authenticated: None, 100 authenticated: None,
89 stdin_tx: None, 101 stdin_tx: None,
90 active_upload: None, 102 active_upload: None,
103 span: Span::none(),
91 } 104 }
92 } 105 }
93 106
@@ -123,7 +136,7 @@ impl SshHandler {
123 } 136 }
124 }, 137 },
125 }, 138 },
126 AuthIdentity::Delegate { certificate } => { 139 AuthIdentity::Delegate { certificate, .. } => {
127 match governance::load(&self.config.repos_dir) { 140 match governance::load(&self.config.repos_dir) {
128 GovernanceState::Active(governance) => { 141 GovernanceState::Active(governance) => {
129 match governance::delegate::validate(certificate, &governance, unix_now()) { 142 match governance::delegate::validate(certificate, &governance, unix_now()) {
@@ -649,6 +662,7 @@ impl Handler for SshHandler {
649 "Public key auth accepted for key type {} ({})", 662 "Public key auth accepted for key type {} ({})",
650 key_type, principal 663 key_type, principal
651 ); 664 );
665 self.span = info_span!("connection", principal = %principal);
652 self.authenticated = Some(AuthIdentity::Key { principal }); 666 self.authenticated = Some(AuthIdentity::Key { principal });
653 Ok(Auth::Accept) 667 Ok(Auth::Accept)
654 } else { 668 } else {
@@ -668,12 +682,16 @@ impl Handler for SshHandler {
668 }; 682 };
669 match governance::delegate::validate(certificate, &governance, unix_now()) { 683 match governance::delegate::validate(certificate, &governance, unix_now()) {
670 Ok(delegate) => { 684 Ok(delegate) => {
671 info!( 685 // The one spelling of a delegate's identity: every log line
672 "Certificate auth accepted: {} (via {})", 686 // this connection goes on to emit carries this exact string,
673 delegate.person, delegate.key_id 687 // so "who did that push" and "who authenticated" are the same
674 ); 688 // question with the same answer.
689 let principal = format!("{} (via {})", delegate.person, delegate.key_id);
690 info!("Certificate auth accepted: {principal}");
691 self.span = info_span!("connection", principal = %principal);
675 self.authenticated = Some(AuthIdentity::Delegate { 692 self.authenticated = Some(AuthIdentity::Delegate {
676 certificate: Box::new(certificate.clone()), 693 certificate: Box::new(certificate.clone()),
694 principal,
677 }); 695 });
678 Ok(Auth::Accept) 696 Ok(Auth::Accept)
679 } 697 }
@@ -690,16 +708,26 @@ impl Handler for SshHandler {
690 reply: ChannelOpenHandle, 708 reply: ChannelOpenHandle,
691 _session: &mut Session, 709 _session: &mut Session,
692 ) -> Result<(), Self::Error> { 710 ) -> Result<(), Self::Error> {
693 if self.authenticated.is_some() { 711 // Every handler body runs inside the connection's span, so identity
694 debug!("Session channel opened: {:?}", channel.id()); 712 // reaches lines that have no principal in hand. `instrument`, not an
695 reply.accept().await; 713 // `entered()` guard: these bodies await, and a guard held across a
696 } else { 714 // yield would leak this connection's identity onto whatever else the
697 warn!("Rejected channel open: not authenticated"); 715 // runtime thread picks up.
698 reply 716 let span = self.span.clone();
699 .reject(russh::ChannelOpenFailure::AdministrativelyProhibited) 717 async move {
700 .await; 718 if self.authenticated.is_some() {
719 debug!("Session channel opened: {:?}", channel.id());
720 reply.accept().await;
721 } else {
722 warn!("Rejected channel open: not authenticated");
723 reply
724 .reject(russh::ChannelOpenFailure::AdministrativelyProhibited)
725 .await;
726 }
727 Ok(())
701 } 728 }
702 Ok(()) 729 .instrument(span)
730 .await
703 } 731 }
704 732
705 async fn exec_request( 733 async fn exec_request(
@@ -708,219 +736,209 @@ impl Handler for SshHandler {
708 data: &[u8], 736 data: &[u8],
709 session: &mut Session, 737 session: &mut Session,
710 ) -> Result<(), Self::Error> { 738 ) -> Result<(), Self::Error> {
711 // russh never answers an exec request for us — not in 0.62, and not in 739 let span = self.span.clone();
712 // 0.46 either, where this call was simply missing. We reply SUCCESS 740 async move {
713 // unconditionally, as sshd does: what the reply reports is whether the 741 // russh never answers an exec request for us — not in 0.62, and not in
714 // *request* was accepted, not whether the command succeeded. The real 742 // 0.46 either, where this call was simply missing. We reply SUCCESS
715 // outcome rides on exit-status, so replying `channel_failure` on a 743 // unconditionally, as sshd does: what the reply reports is whether the
716 // rejected command would make ssh exit 255 with "exec request failed" 744 // *request* was accepted, not whether the command succeeded. The real
717 // and hide those exit codes. 745 // outcome rides on exit-status, so replying `channel_failure` on a
718 // 746 // rejected command would make ssh exit 255 with "exec request failed"
719 // russh drops the packet when the client set `want_reply = false` (it 747 // and hide those exit codes.
720 // gates on `channel.wants_reply`), so this is safe to call blindly. 748 //
721 session.channel_success(channel)?; 749 // russh drops the packet when the client set `want_reply = false` (it
722 750 // gates on `channel.wants_reply`), so this is safe to call blindly.
723 let command_str = match std::str::from_utf8(data) { 751 session.channel_success(channel)?;
724 Ok(s) => s, 752
725 Err(_) => { 753 let command_str = match std::str::from_utf8(data) {
726 warn!("Received non-UTF8 exec request"); 754 Ok(s) => s,
727 session.close(channel)?; 755 Err(_) => {
728 return Ok(()); 756 warn!("Received non-UTF8 exec request");
729 } 757 session.close(channel)?;
730 }; 758 return Ok(());
731 759 }
732 info!("Exec request: {}", command_str); 760 };
733
734 let exec_cmd = match parse_exec_command(command_str) {
735 Some(c) => c,
736 None => {
737 warn!("Rejected exec request: not an allowed command");
738 return reply_and_close(session, channel, "", 1);
739 }
740 };
741
742 let principal = match self.authenticated.as_ref() {
743 Some(AuthIdentity::Key { principal }) => principal.clone(),
744 Some(AuthIdentity::Delegate { certificate }) => {
745 format!("delegate:{}", certificate.key_id())
746 }
747 None => {
748 warn!("Rejected exec request: not authenticated");
749 return reply_and_close(session, channel, "", 1);
750 }
751 };
752
753 let resolved_path = match resolve_repo_path(&self.config.repos_dir, exec_cmd.repo()) {
754 Some(p) => p,
755 None => {
756 warn!("Rejected exec request: path traversal detected");
757 return reply_and_close(session, channel, "error: invalid repo path\n", 1);
758 }
759 };
760
761 let regime = self.regime();
762
763 // The regime is what proves a delegate cert's principal is real and
764 // still enrolled, so only from here on is there a person to name.
765 // Swap the bare-key-ID fallback above for the same "<person> (via
766 // <key-id>)" form the connection-level auth log already uses, so a
767 // delegate's exec/release log lines read the same way.
768 let principal = match &regime {
769 Regime::Governed {
770 name,
771 delegate: Some(key_id),
772 ..
773 } => format!("{name} (via {key_id})"),
774 _ => principal,
775 };
776
777 let git_cmd = match exec_cmd {
778 ExecCommand::Git { cmd, .. } => cmd,
779 ExecCommand::Release(rel) => {
780 return self.handle_release_command(
781 channel,
782 session,
783 rel,
784 &resolved_path,
785 &principal,
786 &regime,
787 );
788 }
789 };
790 761
791 // The name rules are written against, and the repository they name. 762 info!("Exec request: {}", command_str);
792 // Both are needed before the repository exists, because creating one
793 // is itself a permission (`C`).
794 let repo_key = governance::repo_key(&self.config.repos_dir, &resolved_path);
795 let bare;
796 763
797 if resolved_path.exists() { 764 let exec_cmd = match parse_exec_command(command_str) {
798 let entry = match crate::repos::entry_for_path(&self.config.repos_dir, &resolved_path) { 765 Some(c) => c,
799 Some(entry) => entry,
800 None => { 766 None => {
801 warn!( 767 warn!("Rejected exec request: not an allowed command");
802 "Rejected exec request: path is not a git repo: {:?}",
803 resolved_path
804 );
805 return reply_and_close(session, channel, "", 1); 768 return reply_and_close(session, channel, "", 1);
806 } 769 }
807 }; 770 };
808 bare = entry.bare;
809 771
810 // Pushing is authorized here only as far as "may push something"; 772 let principal = match self.authenticated.as_ref() {
811 // which refs may actually move is the update hook's question. 773 Some(AuthIdentity::Key { principal })
812 let needed = match git_cmd { 774 | Some(AuthIdentity::Delegate { principal, .. }) => principal.clone(),
813 GitCmd::UploadPack => Access::Read, 775 None => {
814 GitCmd::ReceivePack => Access::Write, 776 warn!("Rejected exec request: not authenticated");
777 return reply_and_close(session, channel, "", 1);
778 }
815 }; 779 };
816 let authorized = match (&regime, repo_key.as_deref()) { 780
817 (Regime::Closed, _) | (Regime::Governed { .. }, None) => false, 781 let resolved_path = match resolve_repo_path(&self.config.repos_dir, exec_cmd.repo()) {
818 (Regime::Ungoverned, _) => match needed { 782 Some(p) => p,
819 Access::Read => entry.policy.allows_read(&principal), 783 None => {
820 _ => entry.policy.allows_write(&principal), 784 warn!("Rejected exec request: path traversal detected");
821 }, 785 return reply_and_close(session, channel, "error: invalid repo path\n", 1);
822 (
823 Regime::Governed {
824 governance, name, ..
825 },
826 Some(key),
827 ) => {
828 let creator = governance::creator_of(&resolved_path);
829 let subject = Subject::with_creator(name, creator.as_deref());
830 governance.conf.allows_repo(key, &subject, needed)
831 } 786 }
832 }; 787 };
833 788
834 if !authorized { 789 let regime = self.regime();
835 warn!( 790
836 "Rejected exec request: principal {} is not authorized for {} on {:?}", 791 let git_cmd = match exec_cmd {
837 principal, 792 ExecCommand::Git { cmd, .. } => cmd,
838 git_cmd.as_str(), 793 ExecCommand::Release(rel) => {
839 resolved_path 794 return self.handle_release_command(
840 ); 795 channel,
841 return reply_and_close(session, channel, "", 1); 796 session,
842 } 797 rel,
843 } else { 798 &resolved_path,
844 // Auto-creation is what makes wild repos work without a central 799 &principal,
845 // allocator, so under governance it is a permission of its own: 800 &regime,
846 // `C` on a pattern the requested name matches.
847 if let Regime::Governed {
848 governance,
849 name,
850 delegate,
851 } = &regime
852 {
853 let allowed = delegate.is_none()
854 && repo_key.as_deref().is_some_and(|key| {
855 governance
856 .conf
857 .allows_repo(key, &Subject::new(name), Access::Create)
858 });
859 if !allowed {
860 warn!(
861 "Rejected exec request: principal {} may not create {:?}",
862 principal, resolved_path
863 ); 801 );
864 return reply_and_close(session, channel, "", 1);
865 } 802 }
866 } 803 };
867 if matches!(regime, Regime::Closed) {
868 return reply_and_close(session, channel, "", 1);
869 }
870 804
871 match ensure_repo_exists_for_command(git_cmd, &resolved_path) { 805 // The name rules are written against, and the repository they name.
872 Ok(true) => { 806 // Both are needed before the repository exists, because creating one
873 info!("Created bare repo for receive-pack: {:?}", resolved_path); 807 // is itself a permission (`C`).
874 if let Regime::Governed { name, .. } = &regime { 808 let repo_key = governance::repo_key(&self.config.repos_dir, &resolved_path);
875 // Recorded before the push runs, so `RW+ = CREATOR` 809 let bare;
876 // already applies to the very first ref update. 810
877 if let Err(e) = governance::record_creator(&resolved_path, name) { 811 if resolved_path.exists() {
878 error!("Failed to record creator of {:?}: {}", resolved_path, e); 812 let entry =
813 match crate::repos::entry_for_path(&self.config.repos_dir, &resolved_path) {
814 Some(entry) => entry,
815 None => {
816 warn!(
817 "Rejected exec request: path is not a git repo: {:?}",
818 resolved_path
819 );
879 return reply_and_close(session, channel, "", 1); 820 return reply_and_close(session, channel, "", 1);
880 } 821 }
822 };
823 bare = entry.bare;
824
825 // Pushing is authorized here only as far as "may push something";
826 // which refs may actually move is the update hook's question.
827 let needed = match git_cmd {
828 GitCmd::UploadPack => Access::Read,
829 GitCmd::ReceivePack => Access::Write,
830 };
831 let authorized = match (&regime, repo_key.as_deref()) {
832 (Regime::Closed, _) | (Regime::Governed { .. }, None) => false,
833 (Regime::Ungoverned, _) => match needed {
834 Access::Read => entry.policy.allows_read(&principal),
835 _ => entry.policy.allows_write(&principal),
836 },
837 (
838 Regime::Governed {
839 governance, name, ..
840 },
841 Some(key),
842 ) => {
843 let creator = governance::creator_of(&resolved_path);
844 let subject = Subject::with_creator(name, creator.as_deref());
845 governance.conf.allows_repo(key, &subject, needed)
881 } 846 }
882 } 847 };
883 Ok(false) => { 848
849 if !authorized {
884 warn!( 850 warn!(
885 "Rejected exec request: repo path does not exist: {:?}", 851 "Rejected exec request: principal {} is not authorized for {} on {:?}",
852 principal,
853 git_cmd.as_str(),
886 resolved_path 854 resolved_path
887 ); 855 );
888 return reply_and_close(session, channel, "", 1); 856 return reply_and_close(session, channel, "", 1);
889 } 857 }
890 Err(e) => { 858 } else {
891 error!("Failed to create repo {:?}: {}", resolved_path, e); 859 // Auto-creation is what makes wild repos work without a central
860 // allocator, so under governance it is a permission of its own:
861 // `C` on a pattern the requested name matches.
862 if let Regime::Governed {
863 governance,
864 name,
865 delegate,
866 } = &regime
867 {
868 let allowed = delegate.is_none()
869 && repo_key.as_deref().is_some_and(|key| {
870 governance
871 .conf
872 .allows_repo(key, &Subject::new(name), Access::Create)
873 });
874 if !allowed {
875 warn!(
876 "Rejected exec request: principal {} may not create {:?}",
877 principal, resolved_path
878 );
879 return reply_and_close(session, channel, "", 1);
880 }
881 }
882 if matches!(regime, Regime::Closed) {
892 return reply_and_close(session, channel, "", 1); 883 return reply_and_close(session, channel, "", 1);
893 } 884 }
894 }
895 bare = true;
896 }
897 885
898 let child_env = 886 match ensure_repo_exists_for_command(git_cmd, &resolved_path) {
899 match self.receive_pack_env(git_cmd, &regime, repo_key, &resolved_path, bare) { 887 Ok(true) => {
900 Ok(env) => env, 888 info!("Created bare repo for receive-pack: {:?}", resolved_path);
901 Err(reason) => { 889 if let Regime::Governed { name, .. } = &regime {
902 error!("Refusing push to {:?}: {}", resolved_path, reason); 890 // Recorded before the push runs, so `RW+ = CREATOR`
903 return reply_and_close(session, channel, &format!("error: {reason}\n"), 1); 891 // already applies to the very first ref update.
892 if let Err(e) = governance::record_creator(&resolved_path, name) {
893 error!("Failed to record creator of {:?}: {}", resolved_path, e);
894 return reply_and_close(session, channel, "", 1);
895 }
896 }
897 }
898 Ok(false) => {
899 warn!(
900 "Rejected exec request: repo path does not exist: {:?}",
901 resolved_path
902 );
903 return reply_and_close(session, channel, "", 1);
904 }
905 Err(e) => {
906 error!("Failed to create repo {:?}: {}", resolved_path, e);
907 return reply_and_close(session, channel, "", 1);
908 }
904 } 909 }
905 }; 910 bare = true;
906
907 // Create a channel for forwarding client stdin data to the git
908 // subprocess. Tagged with the owning SSH channel so data routing and
909 // EOF teardown can't cross-talk between multiplexed channels.
910 let (tx, rx) = mpsc::channel::<Vec<u8>>(64);
911 self.stdin_tx = Some((channel, tx));
912
913 // Spawn the git subprocess
914 let handle = session.handle();
915 tokio::spawn(async move {
916 if let Err(e) =
917 run_git_command(handle, channel, git_cmd, &resolved_path, child_env, rx).await
918 {
919 error!("Git subprocess error: {}", e);
920 } 911 }
921 });
922 912
923 Ok(()) 913 let child_env =
914 match self.receive_pack_env(git_cmd, &regime, repo_key, &resolved_path, bare) {
915 Ok(env) => env,
916 Err(reason) => {
917 error!("Refusing push to {:?}: {}", resolved_path, reason);
918 return reply_and_close(session, channel, &format!("error: {reason}\n"), 1);
919 }
920 };
921
922 // Create a channel for forwarding client stdin data to the git
923 // subprocess. Tagged with the owning SSH channel so data routing and
924 // EOF teardown can't cross-talk between multiplexed channels.
925 let (tx, rx) = mpsc::channel::<Vec<u8>>(64);
926 self.stdin_tx = Some((channel, tx));
927
928 // Spawn the git subprocess
929 let handle = session.handle();
930 tokio::spawn(async move {
931 if let Err(e) =
932 run_git_command(handle, channel, git_cmd, &resolved_path, child_env, rx).await
933 {
934 error!("Git subprocess error: {}", e);
935 }
936 });
937
938 Ok(())
939 }
940 .instrument(span)
941 .await
924 } 942 }
925 943
926 async fn data( 944 async fn data(
@@ -929,39 +947,44 @@ impl Handler for SshHandler {
929 data: &[u8], 947 data: &[u8],
930 session: &mut Session, 948 session: &mut Session,
931 ) -> Result<(), Self::Error> { 949 ) -> Result<(), Self::Error> {
932 if let Some(upload) = self.active_upload.as_mut() { 950 let span = self.span.clone();
933 if upload.channel == channel { 951 async move {
934 let state = 952 if let Some(upload) = self.active_upload.as_mut() {
935 std::mem::replace(&mut upload.state, UploadState::Failed(String::new())); 953 if upload.channel == channel {
936 upload.state = match state { 954 let state =
937 UploadState::Active(mut active) => match active.write(data) { 955 std::mem::replace(&mut upload.state, UploadState::Failed(String::new()));
938 Ok(()) => UploadState::Active(active), 956 upload.state = match state {
939 // Dropping `active` here discards the temp file. 957 UploadState::Active(mut active) => match active.write(data) {
940 Err(e) => UploadState::Failed(e.to_string()), 958 Ok(()) => UploadState::Active(active),
941 }, 959 // Dropping `active` here discards the temp file.
942 failed => failed, 960 Err(e) => UploadState::Failed(e.to_string()),
943 }; 961 },
944 // Abort as soon as the write fails (e.g. the size cap) rather 962 failed => failed,
945 // than absorbing the rest of the stream and only complaining 963 };
946 // at EOF: tell the client now and close the channel. The 964 // Abort as soon as the write fails (e.g. the size cap) rather
947 // Failed latch stays as a fallback for chunks already in 965 // than absorbing the rest of the stream and only complaining
948 // flight, which then land here with no active upload. 966 // at EOF: tell the client now and close the channel. The
949 if let UploadState::Failed(msg) = &upload.state { 967 // Failed latch stays as a fallback for chunks already in
950 let msg = format!("error: {}\n", msg); 968 // flight, which then land here with no active upload.
951 self.active_upload = None; 969 if let UploadState::Failed(msg) = &upload.state {
952 reply_and_close(session, channel, &msg, 1)?; 970 let msg = format!("error: {}\n", msg);
971 self.active_upload = None;
972 reply_and_close(session, channel, &msg, 1)?;
973 }
974 return Ok(());
953 } 975 }
954 return Ok(());
955 } 976 }
956 } 977 // Forward client data to the git subprocess's stdin, but only for the
957 // Forward client data to the git subprocess's stdin, but only for the 978 // channel that subprocess belongs to.
958 // channel that subprocess belongs to. 979 if let Some((git_channel, tx)) = self.stdin_tx.as_ref() {
959 if let Some((git_channel, tx)) = self.stdin_tx.as_ref() { 980 if *git_channel == channel && tx.send(data.to_vec()).await.is_err() {
960 if *git_channel == channel && tx.send(data.to_vec()).await.is_err() { 981 debug!("stdin channel closed, dropping data");
961 debug!("stdin channel closed, dropping data"); 982 }
962 } 983 }
984 Ok(())
963 } 985 }
964 Ok(()) 986 .instrument(span)
987 .await
965 } 988 }
966 989
967 async fn channel_eof( 990 async fn channel_eof(
@@ -969,28 +992,33 @@ impl Handler for SshHandler {
969 channel: ChannelId, 992 channel: ChannelId,
970 session: &mut Session, 993 session: &mut Session,
971 ) -> Result<(), Self::Error> { 994 ) -> Result<(), Self::Error> {
972 // Close the git subprocess's stdin, but only if this EOF is for the 995 let span = self.span.clone();
973 // channel that owns it. 996 async move {
974 if matches!(self.stdin_tx.as_ref(), Some((git_channel, _)) if *git_channel == channel) { 997 // Close the git subprocess's stdin, but only if this EOF is for the
975 self.stdin_tx = None; 998 // channel that owns it.
976 } 999 if matches!(self.stdin_tx.as_ref(), Some((git_channel, _)) if *git_channel == channel) {
977 1000 self.stdin_tx = None;
978 if let Some(upload) = self.active_upload.take() {
979 if upload.channel != channel {
980 self.active_upload = Some(upload);
981 return Ok(());
982 } 1001 }
983 match upload.state { 1002
984 UploadState::Active(active) => match active.finish() { 1003 if let Some(upload) = self.active_upload.take() {
985 Ok(sha) => reply_and_close(session, channel, &format!("ok {}\n", sha), 0)?, 1004 if upload.channel != channel {
986 Err(e) => reply_and_close(session, channel, &format!("error: {}\n", e), 1)?, 1005 self.active_upload = Some(upload);
987 }, 1006 return Ok(());
988 UploadState::Failed(msg) => { 1007 }
989 reply_and_close(session, channel, &format!("error: {}\n", msg), 1)? 1008 match upload.state {
1009 UploadState::Active(active) => match active.finish() {
1010 Ok(sha) => reply_and_close(session, channel, &format!("ok {}\n", sha), 0)?,
1011 Err(e) => reply_and_close(session, channel, &format!("error: {}\n", e), 1)?,
1012 },
1013 UploadState::Failed(msg) => {
1014 reply_and_close(session, channel, &format!("error: {}\n", msg), 1)?
1015 }
990 } 1016 }
991 } 1017 }
1018 Ok(())
992 } 1019 }
993 Ok(()) 1020 .instrument(span)
1021 .await
994 } 1022 }
995 1023
996 async fn channel_close( 1024 async fn channel_close(
@@ -998,22 +1026,27 @@ impl Handler for SshHandler {
998 channel: ChannelId, 1026 channel: ChannelId,
999 _session: &mut Session, 1027 _session: &mut Session,
1000 ) -> Result<(), Self::Error> { 1028 ) -> Result<(), Self::Error> {
1001 // A close without a preceding EOF still has to release the git child's 1029 let span = self.span.clone();
1002 // stdin, or it stays open until connection teardown. 1030 async move {
1003 if matches!(self.stdin_tx.as_ref(), Some((git_channel, _)) if *git_channel == channel) { 1031 // A close without a preceding EOF still has to release the git child's
1004 self.stdin_tx = None; 1032 // stdin, or it stays open until connection teardown.
1005 } 1033 if matches!(self.stdin_tx.as_ref(), Some((git_channel, _)) if *git_channel == channel) {
1034 self.stdin_tx = None;
1035 }
1006 1036
1007 // A client that closes without EOF gets no reply, but the half-written 1037 // A client that closes without EOF gets no reply, but the half-written
1008 // temp file must still go away. Dropping the ReleaseUpload deletes it. 1038 // temp file must still go away. Dropping the ReleaseUpload deletes it.
1009 if let Some(upload) = self.active_upload.take() { 1039 if let Some(upload) = self.active_upload.take() {
1010 if upload.channel != channel { 1040 if upload.channel != channel {
1011 self.active_upload = Some(upload); 1041 self.active_upload = Some(upload);
1012 } else { 1042 } else {
1013 debug!("Channel closed with an unfinished upload; discarding temp file"); 1043 debug!("Channel closed with an unfinished upload; discarding temp file");
1044 }
1014 } 1045 }
1046 Ok(())
1015 } 1047 }
1016 Ok(()) 1048 .instrument(span)
1049 .await
1017 } 1050 }
1018 } 1051 }
1019 1052
tests/delegate_test.rs
Old New
@@ -124,10 +124,23 @@ fn a_delegate_writes_collab_refs_and_may_not_write_branches() {
124 "delegate collab push failed: {}", 124 "delegate collab push failed: {}",
125 stderr(&push) 125 stderr(&push)
126 ); 126 );
127 // Attribution has to survive past the door. The connection-level
128 // "Certificate auth accepted" line names the delegate once and would
129 // satisfy an assertion over the whole log on its own, so look only at
130 // what follows it, and at a line that names the command being run: that
131 // line can carry an identity only if the command itself is attributed.
132 let log = harness.server_log();
133 let after_auth = log
134 .split_once("Certificate auth accepted")
135 .expect("the certificate should have authenticated")
136 .1;
137 let exec_line = after_auth
138 .lines()
139 .find(|line| line.contains("Exec request"))
140 .unwrap_or_else(|| panic!("the delegate's command should be logged, got: {log}"));
127 assert!( 141 assert!(
128 harness.server_log().contains("claude-a"), 142 exec_line.contains("alex (via claude-a)") && exec_line.contains("git-receive-pack"),
129 "the delegate's key id should be attributed in the server log, got: {}", 143 "the delegate's own command should name who is running it, got: {exec_line}"
130 harness.server_log()
131 ); 144 );
132 145
133 harness 146 harness