…into async-io, as usual.
Showing the changes to git-protocol/src/fetch/blocking_io/mod.rs (1 of 12 files). Show all
Unified · Split
| @@ -0,0 +1,244 @@ | |||
| 1 | /// | ||
| 2 | pub mod arguments; | ||
| 3 | /// | ||
| 4 | pub mod command; | ||
| 5 | /// | ||
| 6 | pub mod delegate; | ||
| 7 | /// | ||
| 8 | pub mod refs; | ||
| 9 | /// | ||
| 10 | pub mod response; | ||
| 11 | #[cfg(test)] | ||
| 12 | mod tests; | ||
| 13 | |||
| 14 | mod error { | ||
| 15 | use crate::fetch::blocking_io::{refs, response}; | ||
| 16 | use git_transport::client; | ||
| 17 | use quick_error::quick_error; | ||
| 18 | use std::io; | ||
| 19 | |||
| 20 | use crate::credentials; | ||
| 21 | |||
| 22 | quick_error! { | ||
| 23 | /// The error used in [`fetch()`]. | ||
| 24 | #[derive(Debug)] | ||
| 25 | #[allow(missing_docs)] | ||
| 26 | pub enum Error { | ||
| 27 | PackIo(err: io::Error) { | ||
| 28 | display("Could not read streaming pack file") | ||
| 29 | from() | ||
| 30 | source(err) | ||
| 31 | } | ||
| 32 | Credentials(err: credentials::Error) { | ||
| 33 | display("Failed to obtain, approve or reject credentials") | ||
| 34 | from() | ||
| 35 | source(err) | ||
| 36 | } | ||
| 37 | Transport(err: client::Error) { | ||
| 38 | display("An error occurred on the transport layer while fetching data") | ||
| 39 | from() | ||
| 40 | source(err) | ||
| 41 | } | ||
| 42 | SymrefWithoutValue { | ||
| 43 | display("A symref 'capability' is expected to have a value") | ||
| 44 | } | ||
| 45 | Ref(err: refs::Error) { | ||
| 46 | display("A reference could not be parsed or invariants were not met") | ||
| 47 | from() | ||
| 48 | source(err) | ||
| 49 | } | ||
| 50 | Response(err: response::Error) { | ||
| 51 | display("The server response could not be parsed") | ||
| 52 | from() | ||
| 53 | source(err) | ||
| 54 | } | ||
| 55 | } | ||
| 56 | } | ||
| 57 | } | ||
| 58 | pub use error::Error; | ||
| 59 | |||
| 60 | mod fetch { | ||
| 61 | use crate::{ | ||
| 62 | credentials, | ||
| 63 | fetch::{ | ||
| 64 | blocking_io::{refs, refs::InternalRef}, | ||
| 65 | Action, Arguments, Command, Delegate, Error, Ref, Response, | ||
| 66 | }, | ||
| 67 | }; | ||
| 68 | use git_features::{progress, progress::Progress}; | ||
| 69 | use git_transport::{ | ||
| 70 | client, | ||
| 71 | client::{SetServiceResponse, TransportV2Ext}, | ||
| 72 | Service, | ||
| 73 | }; | ||
| 74 | use std::io; | ||
| 75 | |||
| 76 | /// Perform a 'fetch' operation with the server using `transport`, with `delegate` handling all server interactions. | ||
| 77 | /// | ||
| 78 | /// * `authenticate(operation_to_perform)` is used to receive credentials for the connection and potentially store it | ||
| 79 | /// if the server indicates 'permission denied'. Note that not all transport support authentication or authorization. | ||
| 80 | /// * `progress` is used to emit progress messages. | ||
| 81 | /// | ||
| 82 | /// _Note_ that depending on the `delegate`, the actual action performed can be `ls-refs`, `clone` or `fetch`. | ||
| 83 | pub fn fetch<F>( | ||
| 84 | mut transport: impl client::Transport, | ||
| 85 | delegate: &mut impl Delegate, | ||
| 86 | mut authenticate: F, | ||
| 87 | mut progress: impl Progress, | ||
| 88 | ) -> Result<(), Error> | ||
| 89 | where | ||
| 90 | F: FnMut(credentials::Action<'_>) -> credentials::Result, | ||
| 91 | { | ||
| 92 | let (protocol_version, mut parsed_refs, capabilities, call_ls_refs) = { | ||
| 93 | progress.init(None, progress::steps()); | ||
| 94 | progress.set_name("handshake"); | ||
| 95 | progress.step(); | ||
| 96 | let result = transport.handshake(Service::UploadPack); | ||
| 97 | let SetServiceResponse { | ||
| 98 | actual_protocol, | ||
| 99 | capabilities, | ||
| 100 | refs, | ||
| 101 | } = match result { | ||
| 102 | Ok(v) => Ok(v), | ||
| 103 | Err(client::Error::Io { ref err }) if err.kind() == io::ErrorKind::PermissionDenied => { | ||
| 104 | drop(result); // needed to workaround this: https://github.com/rust-lang/rust/issues/76149 | ||
| 105 | let url = transport.to_url(); | ||
| 106 | progress.set_name("authentication"); | ||
| 107 | let credentials::Outcome { identity, next } = | ||
| 108 | authenticate(credentials::Action::Fill(&url))?.expect("FILL provides an identity"); | ||
| 109 | transport.set_identity(identity)?; | ||
| 110 | progress.step(); | ||
| 111 | progress.set_name("handshake (authenticated)"); | ||
| 112 | match transport.handshake(Service::UploadPack) { | ||
| 113 | Ok(v) => { | ||
| 114 | authenticate(next.approve())?; | ||
| 115 | Ok(v) | ||
| 116 | } | ||
| 117 | // Still no permission? Reject the credentials. | ||
| 118 | Err(client::Error::Io { err }) if err.kind() == io::ErrorKind::PermissionDenied => { | ||
| 119 | authenticate(next.reject())?; | ||
| 120 | Err(client::Error::Io { err }) | ||
| 121 | } | ||
| 122 | // Otherwise, do nothing, as we don't know if it actually got to try the credentials. | ||
| 123 | // If they were previously stored, they remain. In the worst case, the user has to enter them again | ||
| 124 | // next time they try. | ||
| 125 | Err(err) => Err(err), | ||
| 126 | } | ||
| 127 | } | ||
| 128 | Err(err) => Err(err), | ||
| 129 | }?; | ||
| 130 | |||
| 131 | let mut parsed_refs = Vec::<InternalRef>::new(); | ||
| 132 | refs::from_capabilities(&mut parsed_refs, capabilities.iter())?; | ||
| 133 | |||
| 134 | let call_ls_refs = match refs { | ||
| 135 | Some(mut refs) => { | ||
| 136 | assert_eq!( | ||
| 137 | actual_protocol, | ||
| 138 | git_transport::Protocol::V1, | ||
| 139 | "Only V1 auto-responds with refs" | ||
| 140 | ); | ||
| 141 | refs::from_v1_refs_received_as_part_of_handshake(&mut parsed_refs, &mut refs)?; | ||
| 142 | false | ||
| 143 | } | ||
| 144 | None => true, | ||
| 145 | }; | ||
| 146 | ( | ||
| 147 | actual_protocol, | ||
| 148 | parsed_refs.into_iter().map(Into::into).collect::<Vec<Ref>>(), | ||
| 149 | capabilities, | ||
| 150 | call_ls_refs, | ||
| 151 | ) | ||
| 152 | }; // this scope is needed, see https://github.com/rust-lang/rust/issues/76149 | ||
| 153 | |||
| 154 | if transport.desired_protocol_version() != protocol_version { | ||
| 155 | progress.info(format!( | ||
| 156 | "server did not support protocol {} and downgraded to {}", | ||
| 157 | transport.desired_protocol_version() as usize, | ||
| 158 | protocol_version as usize, | ||
| 159 | )); | ||
| 160 | } | ||
| 161 | |||
| 162 | if call_ls_refs { | ||
| 163 | assert_eq!( | ||
| 164 | protocol_version, | ||
| 165 | git_transport::Protocol::V2, | ||
| 166 | "Only V2 needs a separate request to get specific refs" | ||
| 167 | ); | ||
| 168 | |||
| 169 | let ls_refs = Command::LsRefs; | ||
| 170 | let mut ls_features = ls_refs.default_features(protocol_version, &capabilities); | ||
| 171 | let mut ls_args = ls_refs.initial_arguments(&ls_features); | ||
| 172 | delegate.prepare_ls_refs(&capabilities, &mut ls_args, &mut ls_features); | ||
| 173 | ls_refs.validate_argument_prefixes_or_panic(protocol_version, &capabilities, &ls_args, &ls_features); | ||
| 174 | |||
| 175 | progress.step(); | ||
| 176 | progress.set_name("list refs"); | ||
| 177 | let mut remote_refs = transport.invoke( | ||
| 178 | ls_refs.as_str(), | ||
| 179 | ls_features.into_iter(), | ||
| 180 | if ls_args.is_empty() { | ||
| 181 | None | ||
| 182 | } else { | ||
| 183 | Some(ls_args.into_iter()) | ||
| 184 | }, | ||
| 185 | )?; | ||
| 186 | refs::from_v2_refs(&mut parsed_refs, &mut remote_refs)?; | ||
| 187 | } | ||
| 188 | |||
| 189 | let fetch = Command::Fetch; | ||
| 190 | let mut fetch_features = fetch.default_features(protocol_version, &capabilities); | ||
| 191 | let next = delegate.prepare_fetch(protocol_version, &capabilities, &mut fetch_features, &parsed_refs); | ||
| 192 | fetch.validate_argument_prefixes_or_panic(protocol_version, &capabilities, &[], &fetch_features); | ||
| 193 | |||
| 194 | if next == Action::Close { | ||
| 195 | transport.close()?; | ||
| 196 | return Ok(()); | ||
| 197 | } | ||
| 198 | |||
| 199 | Response::check_required_features(protocol_version, &fetch_features)?; | ||
| 200 | let sideband_all = fetch_features.iter().any(|(n, _)| *n == "sideband-all"); | ||
| 201 | let mut arguments = Arguments::new(protocol_version, fetch_features); | ||
| 202 | let mut previous_response = None::<Response>; | ||
| 203 | let mut round = 1; | ||
| 204 | 'negotiation: loop { | ||
| 205 | progress.step(); | ||
| 206 | progress.set_name(format!("negotiate (round {})", round)); | ||
| 207 | round += 1; | ||
| 208 | let action = delegate.negotiate(&parsed_refs, &mut arguments, previous_response.as_ref()); | ||
| 209 | let mut reader = arguments.send(&mut transport, action == Action::Close)?; | ||
| 210 | if sideband_all { | ||
| 211 | setup_remote_progress(&mut progress, &mut reader); | ||
| 212 | } | ||
| 213 | let response = Response::from_line_reader(protocol_version, &mut reader)?; | ||
| 214 | previous_response = if response.has_pack() { | ||
| 215 | progress.step(); | ||
| 216 | progress.set_name("receiving pack"); | ||
| 217 | if !sideband_all { | ||
| 218 | setup_remote_progress(&mut progress, &mut reader); | ||
| 219 | } | ||
| 220 | delegate.receive_pack(reader, progress, &parsed_refs, &response)?; | ||
| 221 | break 'negotiation; | ||
| 222 | } else { | ||
| 223 | match action { | ||
| 224 | Action::Close => break 'negotiation, | ||
| 225 | Action::Continue => Some(response), | ||
| 226 | } | ||
| 227 | } | ||
| 228 | } | ||
| 229 | Ok(()) | ||
| 230 | } | ||
| 231 | |||
| 232 | fn setup_remote_progress( | ||
| 233 | progress: &mut impl Progress, | ||
| 234 | reader: &mut Box<dyn git_transport::client::ExtendedBufRead + '_>, | ||
| 235 | ) { | ||
| 236 | reader.set_progress_handler(Some(Box::new({ | ||
| 237 | let mut remote_progress = progress.add_child("remote"); | ||
| 238 | move |is_err: bool, data: &[u8]| { | ||
| 239 | crate::RemoteProgress::translate_to_progress(is_err, data, &mut remote_progress) | ||
| 240 | } | ||
| 241 | }) as git_transport::client::HandleProgress)); | ||
| 242 | } | ||
| 243 | } | ||
| 244 | pub use fetch::fetch; | ||