package git-kv

  1. Overview
  2. Docs

Source file smart_git.ml

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
let src = Logs.Src.create "git-fetch"

module SHA1 = Digestif.SHA1
module Log = (val Logs.src_log src : Logs.LOG)
module Pate = Carton_git_lwt.Make (SHA1)

let ( >>? ) = Lwt_result.bind
let ( <.> ) f g = fun x -> f (g x)

let explode store ?threads stream =
  let open Lwt.Infix in
  let buf = Bbuffer.create 0x7ff in
  let append str ~off ~len =
    Log.debug (fun m ->
        m "PACK: @[<hov>%a@]"
          (Hxd_string.pp Hxd.default)
          (String.sub str off len));
    Bbuffer.add_substring buf str off len;
    Lwt.return_unit
  in
  let map buf ~pos:off len =
    let len = Int.min len (Bbuffer.length buf - off) in
    Bbuffer.sub buf off len
  in
  let cachet = Cachet.make ~pagesize:4096 ~map buf in
  let cfg = Pate.config ?threads () in
  Log.debug (fun m -> m "Start to verify the incoming PACK file (and save it)");
  Pate.verify_from_stream ~cfg ~append cachet stream >>= fun (entries, hash) ->
  Log.debug (fun m -> m "Hash of the PACK file: %s" (Ohex.encode hash));
  let offsets = Hashtbl.create 0x7ff in
  let fn = function
    | Carton.Resolved_base {cursor; uid; _} | Resolved_node {cursor; uid; _} ->
      Hashtbl.add offsets uid cursor
    | _ -> ()
  in
  Array.iter fn entries;
  let index (uid : Carton.Uid.t) =
    match Hashtbl.find_opt offsets uid with
    | Some cursor -> Carton.Local cursor
    | None ->
      let hash = SHA1.of_raw_string (uid :> string) in
      let obj = Git_store.read_exn store hash in
      let kind =
        match obj with
        | Git_store.Object.Commit _ -> `A
        | Git_store.Object.Tree _ -> `B
        | Git_store.Object.Blob _ -> `C
        | Git_store.Object.Tag _ -> `D
      in
      let bstr = Git_store.Object.to_bstr obj in
      Carton.Extern (kind, bstr)
  in
  let t = Pate.make ~index cachet in
  let fn = function
    | Carton.Resolved_base {cursor; _}
    | Resolved_node {cursor; _}
    | Unresolved_base {cursor; _}
    | Unresolved_node {cursor} -> (
      let size = Carton.size_of_offset t ~cursor Carton.Size.zero in
      let blob = Carton.Blob.make ~size in
      let value = Carton.of_offset t blob ~cursor in
      let kind =
        match Carton.Value.kind value with
        | `A -> `Commit
        | `B -> `Tree
        | `C -> `Blob
        | `D -> `Tag
      in
      let len = Carton.Value.length value in
      let bstr = Carton.Value.bigstring value in
      let bstr = Bstr.sub bstr ~off:0 ~len in
      match Git_store.Object.of_bstr ~kind bstr with
      | Ok value ->
        let _ = Git_store.write store value in
        ()
      | Error _ -> ())
  in
  Array.iter fn entries; Lwt.return_ok hash

let fetch_v1
    ?(uses_git_transport = false)
    ~push_stdout
    ~push_stderr
    ~capabilities
    path
    flow
    ?deepen
    ?want
    host
    store
    fetch_cfg
    push =
  let open Lwt.Infix in
  Lwt.try_bind
    (fun () ->
      Fetch.fetch_v1 ~uses_git_transport ~push_stdout ~push_stderr ~capabilities
        ?deepen ?want ~host path
        (Smart_flow.Flow.make flow)
        (store, Hashtbl.create 0x7ff)
        fetch_cfg
      @@ fun (payload, off, len) ->
      let v = String.sub payload off len in
      push (Some (v, 0, len)))
    (fun refs ->
      Log.debug (fun m -> m "Got new reference(s) from the server");
      push None >>= fun () ->
      Mimic.close flow >>= fun () -> Lwt.return_ok refs)
  @@ fun exn ->
  push None >>= fun () ->
  Mimic.close flow >>= fun () -> Lwt.reraise exn

let default_capabilities =
  [`Side_band_64k; `Multi_ack_detailed; `Ofs_delta; `Thin_pack; `Report_status]

type transmission = [ `Git | `Exec ]

let rec get_transmission :
    Mimic.edn list ->
    [ `Git | `Exec | `HTTP of Uri.t * Git_store.Endpoint.handshake ] option =
  function
  | Mimic.Edn (k, v) :: r -> (
    match Mimic.equal k Git_store.Endpoint.git_transmission with
    | Some Mimic.Refl -> Some v
    | None -> get_transmission r)
  | [] -> None

let add_unless lst k v =
  match List.assoc_opt (String.lowercase_ascii k) lst with
  | Some _ -> lst
  | None -> (String.lowercase_ascii k, v) :: lst

let pp_version ppf = function
  | `V1 -> Fmt.pf ppf "1"
  | _ -> Fmt.pf ppf "unknown"

let add_headers_for_fetching ?(version = `V1) ctx =
  let headers =
    Option.value ~default:[] (Mimic.get Git_store.Endpoint.git_http_headers ctx)
  in
  let headers =
    add_unless headers "content-type" "application/x-git-upload-pack-request"
  in
  let headers =
    add_unless headers "accept" "application/x-git-upload-pack-result"
  in
  let headers =
    add_unless headers "git-protocol" (Fmt.str "version=%a" pp_version version)
  in
  Mimic.replace Git_store.Endpoint.git_http_headers headers ctx

let fetch
    ?(push_stdout = ignore)
    ?(push_stderr = ignore)
    ?(bounds = 10)
    ?threads
    ~ctx
    store
    edn
    ?(version = `V1)
    ?(capabilities = default_capabilities)
    ?deepen
    want =
  let open Lwt.Infix in
  let hostname = edn.Git_store.Endpoint.hostname in
  let path = edn.Git_store.Endpoint.path in
  let stream, emitter = Lwt_stream.create_bounded bounds in
  let pusher_with_logging = function
    | Some (str, off, len) ->
      Log.debug (fun m -> m "Download %d byte(s) of the PACK file." len);
      emitter#push (String.sub str off len)
    | None ->
      Log.debug (fun m -> m "End of pack.");
      emitter#close;
      Lwt.return_unit
  in
  let ctx =
    Mimic.add Git_store.Endpoint.git_capabilities `Rd
      (Git_store.Endpoint.to_ctx edn ctx)
  in
  let ctx = add_headers_for_fetching ~version ctx in
  Lwt.catch (fun () ->
      Mimic.unfold ctx >>? fun ress ->
      Mimic.connect ress >>= fun flow ->
      match flow, get_transmission ress, version with
      | Ok flow, Some (#transmission as transmission), `V1 -> (
        let fetch_cfg = Fetch.configuration capabilities in
        let uses_git_transport =
          match transmission with `Git -> true | `Exec -> false
        in
        Log.debug (fun m -> m "Start to negotiate and unpack");
        Lwt.both
          (fetch_v1 ~push_stdout ~push_stderr ~uses_git_transport ~capabilities
             path flow ?deepen ~want hostname store fetch_cfg
             pusher_with_logging)
          (explode ?threads store stream)
        >>= fun (refs, idx) ->
        match refs, idx with
        | Ok [], _ -> Lwt.return_ok `Empty
        | Ok refs, Ok uid -> Lwt.return_ok (`Pack (uid, refs))
        | (Error _ as err), _ -> Lwt.return err
        | Ok _refs, (Error _ as err) -> Lwt.return err)
      | Ok flow, Some (`HTTP (uri, handshake)), `V1 -> (
        let fetch_cfg = Fetch.configuration ~stateless:true capabilities in
        let uri0 = Fmt.str "%a/info/refs?service=git-upload-pack" Uri.pp uri in
        let uri0 = Uri.of_string uri0 in
        let uri1 = Fmt.str "%a/git-upload-pack" Uri.pp uri in
        let uri1 = Uri.of_string uri1 in
        Lwt.both
          ( handshake ~uri0 ~uri1 flow >>= fun () ->
            fetch_v1 ~push_stdout ~push_stderr ~capabilities path flow ?deepen
              ~want hostname store fetch_cfg pusher_with_logging )
          (explode ?threads store stream)
        >>= fun (refs, idx) ->
        match refs, idx with
        | Ok refs, Ok uid -> Lwt.return_ok (`Pack (uid, refs))
        | (Error _ as err), _ -> Lwt.return err
        | Ok [], _ -> Lwt.return_ok `Empty
        | Ok _refs, (Error _ as err) -> Lwt.return err)
      | Ok flow, Some _, _ ->
        Log.err (fun m -> m "The protocol version is uninmplemented.");
        Mimic.close flow >>= fun () ->
        Lwt.return_error (`Msg "Version protocol unimplemented")
      | Ok flow, None, _ ->
        Log.err (fun m ->
            m "A flow was allocated but we can not recognize the transmission.");
        Mimic.close flow >>= fun () ->
        Lwt.return_error (`Msg "Unrecognized protocol")
      | Error err, _, _ ->
        Log.err (fun m -> m "The Git peer is not reachable.");
        Lwt.return_error err)
  @@ function
  | Failure err -> Lwt.return_error (`Msg err)
  | exn -> Lwt.return_error (`Exn exn)

let pack store uids =
  let load (uid : Carton.Uid.t) _ =
    let hash = SHA1.of_raw_string (uid :> string) in
    let value = Git_store.read_exn store hash in
    let kind =
      match value with
      | Git_store.Object.Commit _ -> `A
      | Git_store.Object.Tree _ -> `B
      | Git_store.Object.Blob _ -> `C
      | Git_store.Object.Tag _ -> `D
    in
    let bstr = Git_store.Object.to_bstr value in
    Lwt.return (Carton.Value.make ~kind bstr)
  in
  let with_header = List.length uids in
  let uids = Lwt_seq.of_list uids in
  let fn hash =
    let uid = Carton.Uid.unsafe_of_string (SHA1.to_raw_string hash) in
    let value = Git_store.read_exn store hash in
    let kind =
      match value with
      | Git_store.Object.Commit _ -> `A
      | Git_store.Object.Tree _ -> `B
      | Git_store.Object.Blob _ -> `C
      | Git_store.Object.Tag _ -> `D
    in
    let length = Int64.to_int (Git_store.Object.length value) in
    Cartonnage.Entry.make ~kind ~length uid ()
  in
  let entries = Lwt_seq.map fn uids in
  let targets = Pate.delta ~load entries in
  let seq =
    Pate.to_pack ~with_header ~with_signature:SHA1.empty ~load
      (Lwt_stream.of_lwt_seq targets)
  in
  let stream, push = Lwt_stream.create () in
  let open Lwt.Infix in
  Lwt.async (fun () ->
      Lwt_seq.iter (push <.> Option.some) seq >|= fun () -> push None);
  fun () -> Lwt_stream.get stream

let push_v1
    ?uses_git_transport
    flow
    ~capabilities
    path
    cmds
    hostname
    store
    push_cfg
    pack =
  let open Lwt.Infix in
  Push.push ?uses_git_transport ~capabilities cmds ~host:hostname path
    (Smart_flow.Flow.make flow)
    store push_cfg pack
  >>= fun () ->
  Mimic.close flow >>= fun () -> Lwt.return_ok ()

let add_headers_for_pushing ?(version = `V1) ctx =
  let open Git_store.Endpoint in
  let headers = Option.value ~default:[] (Mimic.get git_http_headers ctx) in
  let headers =
    add_unless headers "content-type" "application/x-git-receive-pack-request"
  in
  let headers =
    add_unless headers "accept" "application/x-git-receive-pack-result"
  in
  let headers =
    add_unless headers "git-protocol" (Fmt.str "version=%a" pp_version version)
  in
  Mimic.replace git_http_headers headers ctx

let push
    ~ctx
    ((store, _) as t)
    edn
    ?(version = `V1)
    ?(capabilities = default_capabilities)
    cmds =
  let open Lwt.Infix in
  let hostname = edn.Git_store.Endpoint.hostname in
  let path = edn.Git_store.Endpoint.path in
  let ctx =
    Mimic.add Git_store.Endpoint.git_capabilities `Wr
      (Git_store.Endpoint.to_ctx edn ctx)
  in
  let ctx = add_headers_for_pushing ~version ctx in
  Lwt.catch (fun () ->
      Mimic.unfold ctx >>? fun ress ->
      Mimic.connect ress >>= fun res ->
      match res, get_transmission ress, version with
      | Ok flow, Some (#transmission as transmission), `V1 ->
        let push_cfg = Push.configuration () in
        let uses_git_transport =
          match transmission with `Git -> true | `Exec -> false
        in
        push_v1 ~uses_git_transport flow ~capabilities path cmds hostname t
          push_cfg (pack store)
      | Ok flow, Some (`HTTP (uri, handshake)), `V1 ->
        let push_cfg = Push.configuration ~stateless:true () in
        let uri0 =
          Fmt.str "%a/info/refs?service=git-receive-pack" Uri.pp uri
          |> Uri.of_string
        in
        let uri1 = Fmt.str "%a/git-receive-pack" Uri.pp uri |> Uri.of_string in
        handshake ~uri0 ~uri1 flow >>= fun () ->
        push_v1 flow ~capabilities path cmds hostname t push_cfg (pack store)
      | Ok flow, Some _, _ ->
        Log.err (fun m -> m "The protocol version is uninmplemented.");
        Mimic.close flow >>= fun () ->
        Lwt.return_error (`Msg "Version protocol unimplemented")
      | Ok flow, None, _ ->
        Log.err (fun m ->
            m "A flow was allocated but we can not recognize the transmission.");
        Mimic.close flow >>= fun () ->
        Lwt.return_error (`Msg "Unrecognized protocol")
      | Error err, _, _ ->
        Log.err (fun m -> m "The Git peer is not reachable.");
        Lwt.return_error err)
  @@ function
  | Failure err -> Lwt.return_error (`Msg err)
  | exn -> Lwt.return_error (`Exn exn)
OCaml

Innovation. Community. Security.