dgit.raspbian.org Git - git-annex.git/blob

1 {- proxying

2 -

4 -

5 - Licensed under the GNU AGPL version 3 or higher.

6 -}

7

8 module Annex.Proxy where

9

10 import Annex.Common

11 import qualified Annex

12 import qualified Remote

13 import qualified Types.Remote as Remote

14 import qualified Remote.Git

15 import P2P.Proxy

16 import P2P.Protocol

17 import P2P.IO

18 import Remote.Helper.Ssh (openP2PShellConnection', closeP2PShellConnection)

19 import Annex.Concurrent

20 import Annex.Tmp

21 import Annex.Verify

22 import Annex.UUID

23 import Logs.Proxy

24 import Logs.Cluster

25 import Logs.UUID

26 import Logs.Location

27 import Utility.Tmp.Dir

28 import Utility.Metered

29 import Git.Types

30 import qualified Database.Export as Export

31

32 import Control.Concurrent.STM

33 import Control.Concurrent.Async

34 import qualified Data.ByteString as B

35 import qualified Data.ByteString.Lazy as L

36 import qualified System.FilePath.ByteString as P

37 import qualified Data.Map as M

38 import qualified Data.Set as S

39

40 proxyRemoteSide :: ProtocolVersion -> Bypass -> Remote -> Annex RemoteSide

41 proxyRemoteSide clientmaxversion bypass r

42 | Remote.remotetype r == Remote.Git.remote =

43 proxyGitRemoteSide clientmaxversion bypass r

44 | otherwise =

45 proxySpecialRemoteSide clientmaxversion r

46

47 proxyGitRemoteSide :: ProtocolVersion -> Bypass -> Remote -> Annex RemoteSide

48 proxyGitRemoteSide clientmaxversion bypass r = mkRemoteSide r $

49 openP2PShellConnection' r clientmaxversion bypass >>= \case

50 Just conn@(OpenConnection (remoterunst, remoteconn, _)) ->

51 return $ Just

52 ( remoterunst

53 , remoteconn

54 , void $ liftIO $ closeP2PShellConnection conn

55 )

56 _ -> return Nothing

57

58 proxySpecialRemoteSide :: ProtocolVersion -> Remote -> Annex RemoteSide

59 proxySpecialRemoteSide clientmaxversion r = mkRemoteSide r $ do

60 let protoversion = min clientmaxversion maxProtocolVersion

61 remoterunst <- Serving (Remote.uuid r) Nothing <$>

62 liftIO (newTVarIO protoversion)

63 ihdl <- liftIO newEmptyTMVarIO

64 ohdl <- liftIO newEmptyTMVarIO

65 iwaitv <- liftIO newEmptyTMVarIO

66 owaitv <- liftIO newEmptyTMVarIO

67 iclosedv <- liftIO newEmptyTMVarIO

68 oclosedv <- liftIO newEmptyTMVarIO

69 exportdb <- ifM (Remote.isExportSupported r)

70 ( Just <$> Export.openDb (Remote.uuid r)

71 , pure Nothing

72 )

73 worker <- liftIO . async =<< forkState

74 (proxySpecialRemote protoversion r ihdl ohdl owaitv oclosedv exportdb)

75 let remoteconn = P2PConnection

76 { connRepo = Nothing

77 , connCheckAuth = const False

78 , connIhdl = P2PHandleTMVar ihdl (Just iwaitv) iclosedv

79 , connOhdl = P2PHandleTMVar ohdl (Just owaitv) oclosedv

80 , connIdent = ConnIdent (Just (Remote.name r))

81 }

82 let closeremoteconn = do

83 liftIO $ atomically $ putTMVar oclosedv ()

84 join $ liftIO (wait worker)

85 maybe noop Export.closeDb exportdb

86 return $ Just

87 ( remoterunst

88 , remoteconn

89 , closeremoteconn

90 )

91

92 -- Proxy for the special remote, speaking the P2P protocol.

93 proxySpecialRemote

94 :: ProtocolVersion

95 -> Remote

96 -> TMVar (Either L.ByteString Message)

97 -> TMVar (Either L.ByteString Message)

98 -> TMVar ()

99 -> TMVar ()

100 -> Maybe Export.ExportHandle

101 -> Annex ()

102 proxySpecialRemote protoversion r ihdl ohdl owaitv oclosedv mexportdb = go

103 where

104 go :: Annex ()

105 go = liftIO receivemessage >>= \case

106 Just (CHECKPRESENT k) -> do

107 tryNonAsync (Remote.checkPresent r k) >>= \case

108 Right True -> liftIO $ sendmessage SUCCESS

109 Right False -> liftIO $ sendmessage FAILURE

110 Left err -> liftIO $ propagateerror err

111 go

112 Just (LOCKCONTENT _) -> do

113 -- Special remotes do not support locking content.

114 liftIO $ sendmessage FAILURE

115 go

116 Just (REMOVE k) -> do

117 tryNonAsync (Remote.removeKey r Nothing k) >>= \case

118 Right () -> liftIO $ sendmessage SUCCESS

119 Left err -> liftIO $ propagateerror err

120 go

121 Just (PUT (ProtoAssociatedFile af) k) -> do

122 proxyput af k

123 go

124 Just (GET offset (ProtoAssociatedFile af) k) -> do

125 proxyget offset af k

126 go

127 Just (BYPASS _) -> go

128 Just (CONNECT _) ->

129 -- Not supported and the protocol ends here.

130 liftIO $ sendmessage $ CONNECTDONE (ExitFailure 1)

131 Just NOTIFYCHANGE -> do

132 liftIO $ sendmessage $

133 ERROR "NOTIFYCHANGE unsupported for a special remote"

134 go

135 Just _ -> giveup "protocol error"

136 Nothing -> return ()

137

138 receivemessage = liftIO (atomically recv) >>= \case

139 Right (Right m) -> return (Just m)

140 Right (Left _b) -> giveup "unexpected ByteString received from P2P MVar"

141 Left () -> return Nothing

142 where

143 recv =

144 (Right <$> takeTMVar ohdl)

145 `orElse`

146 (Left <$> readTMVar oclosedv)

147

148 receivebytestring = atomically recv >>= \case

149 Right (Left b) -> return (Just b)

150 Right (Right _m) -> giveup "did not receive ByteString from P2P MVar"

151 Left () -> return Nothing

152 where

153 recv =

154 (Right <$> takeTMVar ohdl)

155 `orElse`

156 (Left <$> readTMVar oclosedv)

157

158 sendmessage m = atomically $ putTMVar ihdl (Right m)

159

160 sendbytestring b = atomically $ putTMVar ihdl (Left b)

161

162 propagateerror err = sendmessage $ ERROR $

163 "proxied special remote reports: " ++ show err

164

165 -- Not using gitAnnexTmpObjectLocation because there might be

166 -- several concurrent GET and PUTs of the same key being proxied

167 -- from this special remote or others, and each needs to happen

168 -- independently. Also, this key is not getting added into the

169 -- local annex objects.

170 withproxytmpfile k a = withOtherTmp $ \othertmpdir ->

171 withTmpDirIn (fromRawFilePath othertmpdir) "proxy" $ \tmpdir ->

172 a (toRawFilePath tmpdir P.</> keyFile k)

173

174 -- Verify the content received from the client, to avoid bad content

175 -- being stored in the special remote.

176 proxyput af k = do

177 liftIO $ sendmessage $ PUT_FROM (Offset 0)

178 withproxytmpfile k $ \tmpfile -> do

179 let store = tryNonAsync (storeput k af (decodeBS tmpfile)) >>= \case

180 Right () -> liftIO $ sendmessage SUCCESS

181 Left err -> liftIO $ propagateerror err

182 liftIO receivemessage >>= \case

183 Just (DATA (Len len)) -> do

184 iv <- startVerifyKeyContentIncrementally Remote.AlwaysVerify k

185 h <- liftIO $ openFile (fromRawFilePath tmpfile) WriteMode

186 gotall <- liftIO $ receivetofile iv h len

187 liftIO $ hClose h

188 verified <- if gotall

189 then fst <$> finishVerifyKeyContentIncrementally' True iv

190 else pure False

191 if protoversion > ProtocolVersion 1

192 then liftIO receivemessage >>= \case

193 Just (VALIDITY Valid)

194 | verified -> store

195 | otherwise -> liftIO $ sendmessage FAILURE

196 Just (VALIDITY Invalid) ->

197 liftIO $ sendmessage FAILURE

198 _ -> giveup "protocol error"

199 else store

200 _ -> giveup "protocol error"

201 liftIO $ removeWhenExistsWith removeFile (fromRawFilePath tmpfile)

202

203 storeput k af tmpfile = case mexportdb of

204 Just exportdb -> liftIO (Export.getExportTree exportdb k) >>= \case

205 [] -> storeputkey k af tmpfile

206 locs -> do

207 havelocs <- liftIO $ S.fromList

208 <$> Export.getExportedLocation exportdb k

209 let locs' = filter (`S.notMember` havelocs) locs

210 forM_ locs' $ \loc ->

211 storeputexport exportdb k loc tmpfile

212 liftIO $ Export.flushDbQueue exportdb

213 Nothing -> storeputkey k af tmpfile

214

215 storeputkey k af tmpfile =

216 Remote.storeKey r k af (Just tmpfile) nullMeterUpdate

217

218 storeputexport exportdb k loc tmpfile = do

219 Remote.storeExport (Remote.exportActions r) tmpfile k loc nullMeterUpdate

220 liftIO $ Export.addExportedLocation exportdb k loc

221

222 receivetofile iv h n = liftIO receivebytestring >>= \case

223 Just b -> do

224 liftIO $ atomically $

225 putTMVar owaitv ()

226 `orElse`

227 readTMVar oclosedv

228 n' <- storetofile iv h n (L.toChunks b)

229 -- Normally all the data is sent in a single

230 -- lazy bytestring. However, when the special

231 -- remote is a node in a cluster, a PUT is

232 -- streamed to it in multiple chunks.

233 if n' == 0

234 then return True

235 else receivetofile iv h n'

236 Nothing -> return False

237

238 storetofile _ _ n [] = pure n

239 storetofile iv h n (b:bs) = do

240 writeVerifyChunk iv h b

241 storetofile iv h (n - fromIntegral (B.length b)) bs

242

243 proxyget offset af k = withproxytmpfile k $ \tmpfile -> do

244 -- Don't verify the content from the remote,

245 -- because the client will do its own verification.

246 let vc = Remote.NoVerify

247 tryNonAsync (Remote.retrieveKeyFile r k af (fromRawFilePath tmpfile) nullMeterUpdate vc) >>= \case

248 Right _ -> liftIO $ senddata offset tmpfile

249 Left err -> liftIO $ propagateerror err

250

251 senddata (Offset offset) f = do

252 size <- fromIntegral <$> getFileSize f

253 let n = max 0 (size - offset)

254 sendmessage $ DATA (Len n)

255 withBinaryFile (fromRawFilePath f) ReadMode $ \h -> do

256 hSeek h AbsoluteSeek offset

257 sendbs =<< L.hGetContents h

258 -- Important to keep the handle open until

259 -- the client responds. The bytestring

260 -- could still be lazily streaming out to

261 -- the client.

262 waitclientresponse

263 where

264 sendbs bs = do

265 sendbytestring bs

266 when (protoversion > ProtocolVersion 0) $

267 sendmessage (VALIDITY Valid)

268

269 waitclientresponse =

270 receivemessage >>= \case

271 Just SUCCESS -> return ()

272 Just FAILURE -> return ()

273 Just _ -> giveup "protocol error"

274 Nothing -> return ()

275

276 {- Check if this repository can proxy for a specified remote uuid,

277 - and if so enable proxying for it. -}

278 checkCanProxy :: UUID -> UUID -> Annex Bool

279 checkCanProxy remoteuuid myuuid = do

280 myproxies <- M.lookup myuuid <$> getProxies

281 checkCanProxy' myproxies remoteuuid >>= \case

282 Right v -> do

283 Annex.changeState $ \st -> st { Annex.proxyremote = Just v }

284 return True

285 Left Nothing -> return False

286 Left (Just err) -> giveup err

287

288 checkCanProxy' :: Maybe (S.Set Proxy) -> UUID -> Annex (Either (Maybe String) (Either ClusterUUID Remote))

289 checkCanProxy' Nothing _ = return (Left Nothing)

290 checkCanProxy' (Just proxies) remoteuuid =

291 case filter (\p -> proxyRemoteUUID p == remoteuuid) (S.toList proxies) of

292 [] -> notconfigured

293 ps -> case mkClusterUUID remoteuuid of

294 Just cu -> proxyforcluster cu

295 Nothing -> proxyfor ps

296 where

297 proxyfor ps = do

298 rs <- concat . Remote.byCost <$> Remote.remoteList

299 myclusters <- annexClusters <$> Annex.getGitConfig

300 case canProxyForRemote rs ps myclusters remoteuuid of

301 Nothing -> notconfigured

302 Just r -> return (Right (Right r))

303

304 proxyforcluster cu = do

305 clusters <- getClusters

306 if M.member cu (clusterUUIDs clusters)

307 then return (Right (Left cu))

308 else notconfigured

309

310 notconfigured = M.lookup remoteuuid <$> uuidDescMap >>= \case

311 Just desc -> return $ Left $ Just $

312 "not configured to proxy for repository " ++ fromUUIDDesc desc

313 Nothing -> return $ Left Nothing

314

315 {- Remotes that this repository is configured to proxy for.

316 -

317 - When there are multiple remotes that access the same repository,

318 - this picks the lowest cost one that is configured to be used as a proxy.

319 -}

320 proxyForRemotes :: Annex [Remote]

321 proxyForRemotes = do

322 myuuid <- getUUID

323 (M.lookup myuuid <$> getProxies) >>= \case

324 Nothing -> return []

325 Just myproxies -> do

326 let myproxies' = S.toList myproxies

327 rs <- concat . Remote.byCost <$> Remote.remoteList

328 myclusters <- annexClusters <$> Annex.getGitConfig

329 return $ mapMaybe (canProxyForRemote rs myproxies' myclusters . Remote.uuid) rs

330

331 -- Only proxy for a remote when the git configuration allows it.

332 -- This is important to prevent changes to the git-annex branch

333 -- causing unexpected proxying for remotes.

334 canProxyForRemote

335 :: [Remote] -- ^ must be sorted by cost

336 -> [Proxy]

337 -> M.Map RemoteName ClusterUUID

338 -> UUID

339 -> (Maybe Remote)

340 canProxyForRemote rs myproxies myclusters remoteuuid =

341 headMaybe $ filter canproxy rs

342 where

343 canproxy r =

344 sameuuid r &&

345 proxyisconfigured r &&

346 any (isproxyfor r) myproxies

347

348 sameuuid r = Remote.uuid r == remoteuuid

349

350 isproxyfor r p =

351 proxyRemoteUUID p == remoteuuid &&

352 Remote.name r == proxyRemoteName p

353

354 proxyisconfigured r

355 | remoteAnnexProxy (Remote.gitconfig r) = True

356 -- Proxy for remotes that are configured as cluster nodes.

357 | any (`M.member` myclusters) (fromMaybe [] $ remoteAnnexClusterNode $ Remote.gitconfig r) = True

358 -- Proxy for a remote when it is proxied by another remote

359 -- which is itself configured as a cluster gateway.

360 | otherwise = case remoteAnnexProxiedBy (Remote.gitconfig r) of

361 Just proxyuuid -> not $ null $

362 concatMap (remoteAnnexClusterGateway . Remote.gitconfig) $

363 filter (\p -> Remote.uuid p == proxyuuid) rs

364 Nothing -> False

365

366 mkProxyMethods :: ProxyMethods

367 mkProxyMethods = ProxyMethods

368 { removedContent = \lu u k -> logChange lu k u InfoMissing

369 , addedContent = \lu u k -> logChange lu k u InfoPresent

370 }