diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,25 @@ and this project adheres to the [Haskell Package Versioning Policy](https://pvp.haskell.org/). +## [1.2.0] - 2026-11-08 + +### Security +- **Gallery downloader**: file names from the server could contain path separators, letting a forged list write anywhere the client can. Lists must now carry a valid `DLHASH`, and names/types are limited to what the gallery system produces (official 1.6.6 fix) + +### Changed +- **Protocol 1.6.6 (build 181)**: RPC moves to `/16/rpc` and `/16/dl`; replies must be signed with the client key and unsigned or altered ones are rejected. The build is fixed in code; the config `version` field is no longer used +- **Flood control** now matches the official client (10 hits draining 1/s, 60 s block; local network and RPC servers exempt) +- **Cache misses** are streamed to the requester while being fetched instead of after the full download +- **Hackage**: builds without extra flags; `cryptostore` is vendored with crypton baked in + +### Fixed +- `threaded_proxy_test` reports real results; `servercmd` signatures are checked +- Files whose hash is `unknown` (generated on demand) can now be downloaded +- webp/avif Content-Type; a failed cache write no longer fails the request; shutdown no longer spins when the RPC server is unreachable + +### Added +- `hath-bench`: a mock RPC server and benchmark that drives hs-hath or the official client + ## [1.1.2] - 2026-01-31 ### Changed diff --git a/app/Bench.hs b/app/Bench.hs new file mode 100644 --- /dev/null +++ b/app/Bench.hs @@ -0,0 +1,425 @@ +{-# LANGUAGE BangPatterns #-} + +-- | hath-bench: plays the H@H RPC server for a client running on this machine +-- (hs-hath or the official Java client), replays the measurements the real +-- server makes, and prints a scorecard. +-- +-- The real server's quality inputs we can reproduce: +-- * servercmd speed_test / still_alive right after client_start +-- * servercmd threaded_proxy_test (the client measures someone else) +-- * other clients pulling /t/ (5 x 5 MB concurrently) +-- * readers pulling /h/ files, cold (srfetch + upstream) and warm +module Bench ( main ) where + +import qualified Data.ByteString as BS +import qualified Data.ByteString.Builder as BB +import qualified Data.ByteString.Char8 as BSC +import qualified Data.ByteString.Lazy as LBS +import qualified Data.List as List +import qualified Data.Map.Strict as Map +import Data.String.Interpolate ( i ) +import qualified Data.Text as T +import Data.Time.Clock.System ( getSystemTime, systemSeconds ) + +import GHC.Clock ( getMonotonicTime ) + +import Network.Connection ( TLSSettings(TLSSettingsSimple) ) +import qualified Network.HTTP.Client as HC +import Network.HTTP.Client.TLS ( mkManagerSettings ) +import qualified Network.HTTP.Types as H +import Network.TLS ( defaultSupported ) +import qualified Network.Wai as Wai +import qualified Network.Wai.Handler.Warp as Warp + +import Options.Applicative + +import Relude + +import System.Directory ( createDirectoryIfMissing, makeAbsolute ) +import System.Environment ( getEnvironment ) +import System.FilePath ( () ) +import System.Process ( CreateProcess(env), proc, readCreateProcess ) + +import Text.Printf ( printf ) + +import Control.Concurrent ( forkIO, threadDelay ) + +import Control.Exception ( try ) + +import UnliftIO ( pooledMapConcurrentlyN, timeout ) + +import Utils ( hash ) + +data Opts + = Opts + { oClientId :: !Text + , oClientKey :: !Text + , oRpcPort :: !Int + , oClientPort :: !Int + , oFiles :: !Int + , oFileSize :: !Int + , oUpDelayMs :: !Int + , oUpKBps :: !Int + , oWorkDir :: !FilePath + , oClientHost :: !Text -- ^ where the bench reaches the client + , oPublicHost :: !Text -- ^ where the client reaches the bench (also its rpc_server_ip) + } + +optsP :: Parser Opts +optsP + = Opts + <$> strOption (long "client-id" <> value "12345" <> showDefault <> help "Client ID of the client under test") + <*> strOption + (long "client-key" + <> value "benchbenchbenchbench" + <> showDefault + <> help "Client key (the Java client wants 20 alphanumerics)") + <*> option auto + (long "rpc-port" + <> value 19800 + <> showDefault + <> help "Port of the fake RPC server, image server and speed-test target") + <*> option auto (long "client-port" <> value 18443 <> showDefault <> help "Port the client is told to listen on") + <*> option auto (long "files" <> value 30 <> showDefault <> help "Gallery size for the /h/ phases") + <*> option auto (long "file-size" <> value 300000 <> showDefault <> help "Bytes per gallery file") + <*> option auto + (long "upstream-delay-ms" + <> value 300 + <> showDefault + <> help "Latency the fake image server adds before its first byte") + <*> option auto + (long "upstream-kbps" + <> value 1000 + <> showDefault + <> help "Fake image server bandwidth in KB/s (0 = unlimited)") + <*> strOption (long "work-dir" <> value "bench-work" <> showDefault <> help "Where certs and client configs go") + <*> strOption (long "client-host" <> value "127.0.0.1" <> showDefault <> help "Address of the client under test") + <*> strOption + (long "public-host" + <> value "127.0.0.1" + <> showDefault + <> help "Address the client uses to reach this bench; anything but 127.0.0.1 listens on all IPv4 interfaces") + +data GFile = GFile { gIndex :: !Int, gId :: !Text, gBytes :: !ByteString } + +data Bench + = Bench + { bOpts :: !Opts + , bP12 :: !ByteString + , bFiles :: !(Map Text GFile) + , bRpcCalls :: !(IORef (Map Text Int)) + , bSrfetch :: !(IORef Int) + , bStarted :: !(MVar ()) + } + +main :: IO () +main = do + hSetBuffering stdout LineBuffering + o <- execParser + (info (optsP <**> helper) (fullDesc <> progDesc "Fake H@H RPC server that measures a local client")) + when (oFileSize o < 64) $ die "--file-size must be at least 64" + wd <- makeAbsolute (oWorkDir o) + createDirectoryIfMissing True (wd "java" "data") + p12 <- makeCert o wd + writeClientConfigs o wd + -- a fresh nonce per run gives new file ids, so the cold phase is really cold + nonce <- systemSeconds <$> getSystemTime + let files = mkGallery nonce (oFiles o) (oFileSize o) + b <- Bench o p12 (Map.fromList [ ( gId f, f ) | f <- files ]) + <$> newIORef mempty + <*> newIORef 0 + <*> newEmptyMVar + void + $ forkIO + $ Warp.runSettings + (Warp.setHost (if oPublicHost o == "127.0.0.1" then "127.0.0.1" else "*4") + $ Warp.setPort (oRpcPort o) Warp.defaultSettings) + (app b) + putTextLn (instructions o wd) + takeMVar (bStarted b) + -- the Java client only accepts normal connections once client_start has returned + putTextLn "client_start received, starting in 3 s" + threadDelay 3000000 + mgr <- HC.newManager + (mkManagerSettings (TLSSettingsSimple True False False defaultSupported) Nothing) + { HC.managerResponseTimeout = HC.responseTimeoutMicro (90 * 1000000) } + rows <- runPhases b mgr files + calls <- readIORef (bRpcCalls b) + report o rows calls + unless (all rowPassed rows) exitFailure + +-- | Self-signed cert packed as PKCS#12 with the client key as password, which is what get_cert returns. +makeCert :: Opts -> FilePath -> IO ByteString +makeCert o wd = do + environment <- getEnvironment + let openssl args = void + $ readCreateProcess + (proc "openssl" args) + { env = Just (( "HATH_BENCH_PASS", toString (oClientKey o) ) : environment) } + "" + openssl + [ "req", "-x509", "-newkey", "rsa:2048", "-nodes", "-days", "7", "-subj", "/CN=127.0.0.1" + , "-keyout", wd "key.pem", "-out", wd "cert.pem" ] + openssl + [ "pkcs12", "-export", "-inkey", wd "key.pem", "-in", wd "cert.pem" + , "-out", wd "cert.p12", "-passout", "env:HATH_BENCH_PASS" + , "-name", "hath.network" ] -- the alias the Java client looks up + readFileBS (wd "cert.p12") + +writeClientConfigs :: Opts -> FilePath -> IO () +writeClientConfigs o wd = do + writeFileText (wd "java" "data" "client_login") [i|#{oClientId o}-#{oClientKey o}|] + writeFileText + (wd "hs-hath.dhall") + [i|let CacheBackend = < CacheBackendSQLite | CacheBackendR2 | CacheBackendFilesystem > +let ClientProxy = { host : Text, port : Integer, auth : Optional { _1 : Text, _2 : Text } } +in { clientId = "#{oClientId o}" + , key = "#{oClientKey o}" + , version = "176" + , proxy = None ClientProxy + , downloadDir = "#{wd}/hs-download" + , cachePath = "#{wd}/hs-cache.db" + , cacheBackend = CacheBackend.CacheBackendSQLite + , r2Config = None { r2Endpoint : Text, r2Bucket : Text } + , lruCacheSize = +104857600 + } +|] + +instructions :: Opts -> FilePath -> Text +instructions o wd + = [i|hath-bench: fake RPC server on #{oPublicHost o}:#{oRpcPort o}. Start the client under test on #{oClientHost o} +(copy #{wd} there first if that is another machine): + + hs-hath: cd #{wd} && HATH_RPC_HOST=#{oPublicHost o}:#{oRpcPort o} hs-hath --config ./hs-hath.dhall --skip-startup-verify + Java: cd #{wd}/java && java -jar /path/to/HentaiAtHome.jar --rpc_server_ip=#{oPublicHost o} --rpc_server_port=#{oRpcPort o} + +(both configs use client id #{oClientId o}; pass --client-id/--client-key to change them) +Waiting for client_start...|] + +mkGallery :: Int64 -> Int -> Int -> [ GFile ] +mkGallery nonce n size + = [ GFile k fileId body + | k <- [ 1 .. n ] + , let prefix = encodeUtf8 ([i|hath-bench #{nonce} #{k} |] :: Text) + body = prefix <> BS.replicate (size - BS.length prefix) 46 + fileId = [i|#{hash @Text body}-#{size}-1280-1707-jpg|] ] + +-- | RPC endpoint, fake image server (/up) and speed-test target (/t) on one port. +app :: Bench -> Wai.Application +app b req respond = case Wai.pathInfo req of + [ "15", "rpc" ] -> rpc False + -- 1.6.6+ (build 181): same actions, signed responses + [ "16", "rpc" ] -> rpc True + [ "up", fileId ] -> upstream fileId + [ "t", sizeText, _, testKey, _ ] -> testTarget sizeText testKey + _ -> respond $ Wai.responseLBS H.status404 [] "" + where + o = bOpts b + + -- Not Wai.queryString: parseQuery also splits on ';', and the Java client sends + -- add=fileindex;xres;fileid unescaped. + param k + = maybe "" (H.urlDecode True . BS.drop (BS.length k + 1)) + $ find ((k <> "=") `BS.isPrefixOf`) + $ BSC.split '&' (BS.dropWhile (== 63) (Wai.rawQueryString req)) + + -- hs-hath's servant client insists on the real server's content type, + -- the Java client on a Content-Length + rpcResponse body + = respond + $ Wai.responseLBS + H.status200 + [ ( H.hContentType, "text/plain;charset=UTF-8" ), ( H.hContentLength, show (BS.length body) ) ] + (LBS.fromStrict body) + + -- 1.6.6 signs replies: OK---" ++ concatMap ("\n" ++) lines)> + okWith signed ls = do + now <- systemSeconds <$> getSystemTime + let sig = hash @Text (encodeUtf8 (T.concat ([i|#{now}-#{oClientKey o}|] : map ("\n" <>) ls))) + status = if signed then [i|OK-#{now}-#{sig}|] else "OK" + rpcResponse (encodeUtf8 (T.unlines (status : ls))) + + settings + = [ [i|rpc_server_ip=#{oPublicHost o}|] + , [i|host=#{oClientHost o}|] + , [i|port=#{oClientPort o}|] + -- what the real server sends; Java stalls non-local transfers on throttle_bytes=0 + , "throttle_bytes=640000000" + , "disable_bwm=true" + , "disklimit_bytes=1073741824" + , "diskremaining_bytes=1073741824" ] + + rpc signed = do + let ok = okWith signed + let act = decodeUtf8 (param "act") :: Text + atomicModifyIORef' (bRpcCalls b) (\m -> ( Map.insertWith (+) act 1 m, () )) + now <- systemSeconds <$> getSystemTime + case act of + "server_stat" -> ok [ "min_client_build=1", "cur_client_build=999", [i|server_time=#{now}|] ] + "client_login" -> ok settings + "client_settings" -> ok settings + "get_cert" -> rpcResponse (bP12 b) + "client_start" -> tryPutMVar (bStarted b) () >> ok [] + "srfetch" -> do + atomicModifyIORef' (bSrfetch b) (\n -> ( n + 1, () )) + -- add = fileindex;xres;fileid + let fileId = decodeUtf8 (BSC.takeWhileEnd (/= ';') (param "add")) :: Text + ok [ [i|http://#{oPublicHost o}:#{oRpcPort o}/up/#{fileId}|] ] + _ -> ok [] -- still_alive, get_blacklist, client_stop, ... + + upstream fileId = case Map.lookup fileId (bFiles b) of + Nothing -> respond $ Wai.responseLBS H.status404 [] "" + Just f -> do + threadDelay (oUpDelayMs o * 1000) + let body = gBytes f + respond + $ Wai.responseStream + H.status200 + [ ( H.hContentType, "image/jpeg" ), ( H.hContentLength, show (BS.length body) ) ] + $ \write flush -> forM_ (chunksOf 16384 body) $ \c -> do + write (BB.byteString c) >> flush + when (oUpKBps o > 0) $ threadDelay (BS.length c * 1000 `div` oUpKBps o) + + testTarget sizeText testKey = case readMaybe @Int (toString sizeText) of + Nothing -> respond $ Wai.responseLBS H.status400 [] "" + Just size + | testKey == "broken403" -> respond $ Wai.responseLBS H.status403 [] "" + -- no Content-Length and only half the bytes: the tester sees a clean, early EOF + | testKey == "brokenshort" -> zeros [] (size `div` 2) + | otherwise -> zeros [ ( H.hContentLength, show size ) ] size + + zeros headers n + = respond + $ Wai.responseStream H.status200 headers + $ \write _ -> forM_ (chunksOf 65536 (BS.replicate n 48)) (write . BB.byteString) + +chunksOf :: Int -> ByteString -> [ ByteString ] +chunksOf n bs + | BS.null bs = [] + | otherwise = let ( a, rest ) = BS.splitAt n bs in a : chunksOf n rest + +data Sample + = Sample + { sStatus :: !(Maybe Int) -- ^ Nothing: connection error or timeout + , sTtfb :: !Double -- ^ seconds until response headers + , sTotal :: !Double -- ^ seconds until the body is drained + , sBytes :: !Int + , sHead :: !ByteString -- ^ first bytes of the body + } + +-- | GET with redirects followed, timing headers and body separately. +fetch :: HC.Manager -> String -> IO Sample +fetch mgr url = do + t0 <- getMonotonicTime + r <- timeout (90 * 1000000) $ try @SomeException $ do + req <- HC.parseRequest url + HC.withResponse req mgr $ \resp -> do + t1 <- getMonotonicTime + let drain !n acc = HC.brRead (HC.responseBody resp) >>= \c -> if BS.null c + then pure ( n, acc ) + else drain (n + BS.length c) (if BS.length acc < 256 then acc <> BS.take 256 c else acc) + ( n, headBytes ) <- drain 0 "" + t2 <- getMonotonicTime + pure $ Sample (Just (H.statusCode (HC.responseStatus resp))) (t1 - t0) (t2 - t0) n headBytes + t3 <- getMonotonicTime + pure $ case r of + Just (Right s) -> s + _ -> Sample Nothing (t3 - t0) (t3 - t0) 0 "" + +data Row + = Row + { rowName :: !Text + , rowSamples :: ![ Sample ] + , rowWall :: !Double + , rowPassed :: !Bool + , rowNote :: !Text + } + +runPhases :: Bench -> HC.Manager -> [ GFile ] -> IO [ Row ] +runPhases b mgr files = do + now <- systemSeconds <$> getSystemTime + let o = bOpts b + base = [i|https://#{oClientHost o}:#{oClientPort o}|] :: String + sha1 = hash @Text . encodeUtf8 @Text + + servercmd :: Text -> Text -> String + servercmd cmd add + = let + sig = sha1 [i|hentai@home-servercmd-#{cmd}-#{add}-#{oClientId o}-#{now}-#{oClientKey o}|] + in + [i|#{base}/servercmd/#{cmd}/#{add}/#{now}/#{sig}|] + + proxyTest :: Text -> String + proxyTest k + = servercmd + "threaded_proxy_test" + [i|hostname=#{oPublicHost o};protocol=http;port=#{oRpcPort o};testsize=1000000;testcount=5;testtime=#{now};testkey=#{k}|] + + speedKey = sha1 [i|hentai@home-speedtest-5000000-#{now}-#{oClientId o}-#{oClientKey o}|] + + fileUrl f + = let + stamp = T.take 10 (sha1 [i|#{now}-#{gId f}-#{oClientKey o}-hotlinkthis|]) + in + [i|#{base}/h/#{gId f}/keystamp=#{now}-#{stamp};fileindex=#{gIndex f};xres=org/#{gIndex f}.jpg|] + + -- run n requests with the given concurrency; the check sees all samples + phase name conc urls check = do + t0 <- getMonotonicTime + ss <- pooledMapConcurrentlyN conc (fetch mgr) urls + t1 <- getMonotonicTime + let ( passed, note ) = check ss + pure $ Row name ss (t1 - t0) passed note + + allFull size ss = all (\s -> sStatus s == Just 200 && sBytes s == size) ss + bodyText = decodeUtf8 . sHead :: Sample -> Text + bodyIs p ss = ( all (\s -> sStatus s == Just 200 && p (bodyText s)) ss, T.intercalate " " (bodyText <$> ss) ) + + galleryPhase name expectFetches = do + before <- readIORef (bSrfetch b) + row <- phase name 4 (fileUrl <$> files) (\ss -> ( allFull (oFileSize o) ss, "" )) + fetches <- subtract before <$> readIORef (bSrfetch b) + pure row { rowPassed = rowPassed row && fetches == expectFetches + , rowNote = [i|srfetch=#{fetches} (want #{expectFetches})|] } + + -- for poking the client from a third machine (e.g. a flood-control burst) + writeFileText (oWorkDir o "urls.txt") (T.unlines (toText . fileUrl <$> files)) + sequence + [ phase "servercmd still_alive" 1 [ servercmd "still_alive" "" ] (bodyIs (const True)) + , phase "servercmd speed_test" 2 (replicate 2 (servercmd "speed_test" "testsize=2500000")) + (\ss -> ( allFull 2500000 ss, "" )) + , phase "servercmd unsigned" 1 [ [i|#{base}/servercmd/still_alive//#{now}/0|] ] + (\ss -> ( map sStatus ss == [ Just 403 ], "want 403" )) + , phase "proxy_test healthy" 1 [ proxyTest "ok" ] (bodyIs ("OK:5-" `T.isPrefixOf`)) + , phase "proxy_test target 403" 1 [ proxyTest "broken403" ] (bodyIs (== "OK:0-0")) + , phase "proxy_test short body" 1 [ proxyTest "brokenshort" ] (bodyIs (== "OK:0-0")) + , phase "/t/ 5x5MB concurrent" 5 + [ [i|#{base}/t/5000000/#{now}/#{speedKey}/#{k}|] | k <- [ 1 .. 5 :: Int ] ] + (\ss -> ( allFull 5000000 ss, "" )) + , galleryPhase "/h/ cold (cache miss)" (length files) + , galleryPhase "/h/ warm (cache hit)" 0 + ] + +report :: Opts -> [ Row ] -> Map Text Int -> IO () +report o rows calls = do + printf "\nhath-bench results (client on %s:%d)\n\n" (toString (oClientHost o)) (oClientPort o) + printf "%-24s %-14s %15s %15s %8s %s\n" + ("phase" :: String) ("status" :: String) ("ttfb p50/p95" :: String) + ("total p50/p95" :: String) ("MB/s" :: String) ("check" :: String) + forM_ rows $ \r -> do + let ss = rowSamples r + statuses = T.unwords + [ [i|#{maybe "err" (show @Text) s}x#{c}|] :: Text + | ( s, c ) <- Map.toList (Map.fromListWith (+) [ ( sStatus x, 1 :: Int ) | x <- ss ]) ] + ms f = [i|#{msAt 0.5 f ss}/#{msAt 0.95 f ss}|] :: Text + mbps = fromIntegral (sum (sBytes <$> ss)) / max 1e-9 (rowWall r) / 1e6 :: Double + printf "%-24s %-14s %15s %15s %8.1f %s %s\n" + (toString (rowName r)) (toString statuses) (toString (ms sTtfb)) (toString (ms sTotal)) mbps + (if rowPassed r then "ok" else "FAIL" :: String) (toString (rowNote r)) + printf "\nRPC calls seen: %s\n" + (toString (T.unwords [ [i|#{a}x#{n}|] | ( a, n ) <- Map.toList calls ])) + where + msAt :: Double -> (Sample -> Double) -> [ Sample ] -> Int + msAt p f ss = case sort (f <$> ss) of + [] -> 0 + xs -> round (1000 * (xs List.!! min (length xs - 1) (floor (p * fromIntegral (length xs))))) diff --git a/app/Migrate.hs b/app/Migrate.hs --- a/app/Migrate.hs +++ b/app/Migrate.hs @@ -42,7 +42,7 @@ import System.Directory ( doesDirectoryExist, doesFileExist, listDirectory ) import System.FilePath ( (), takeFileName ) -import Types ( FileURI, R2Config(..), RPCError(..), parseFileURI ) +import Types ( FileURI, R2Config(..), RPCError(..), parseFileURI, renderFileURIText ) -- | Command line options data MigrateOptions @@ -274,6 +274,6 @@ fileURIToKey :: FileURI -> Text fileURIToKey uri = s4 <> "/" <> fileId where - fileId = show uri + fileId = renderFileURIText uri s4 = T.take 4 fileId diff --git a/hs-hath.cabal b/hs-hath.cabal --- a/hs-hath.cabal +++ b/hs-hath.cabal @@ -5,7 +5,7 @@ -- see: https://github.com/sol/hpack name: hs-hath -version: 1.1.2 +version: 1.2.0 synopsis: A Haskell implementation of the Hentai@Home client description: Please see the README on GitHub at category: Web @@ -85,13 +85,13 @@ , crypton-connection ==0.4.* , crypton-x509 ==1.7.* , crypton-x509-store ==1.6.* - , cryptostore ==0.3.* , dhall ==1.42.* , directory ==1.3.* , exceptions ==0.10.* , extra >=1.7 && <1.9 , filepath >=1.4 && <1.6 , hex-text ==0.1.* + , hs-hath-cryptostore , hs-hath-minio , hs-hath-placeholder , hs-hath-polysemy-zoo @@ -137,6 +137,113 @@ , warp ==3.4.* default-language: GHC2021 +library hs-hath-cryptostore + exposed-modules: + Crypto.Store.ASN1.Generate + Crypto.Store.ASN1.Parse + Crypto.Store.Cipher.RC2 + Crypto.Store.Cipher.RC2.Primitive + Crypto.Store.CMS + Crypto.Store.CMS.Algorithms + Crypto.Store.CMS.Attribute + Crypto.Store.CMS.Authenticated + Crypto.Store.CMS.AuthEnveloped + Crypto.Store.CMS.Digested + Crypto.Store.CMS.Encrypted + Crypto.Store.CMS.Enveloped + Crypto.Store.CMS.Info + Crypto.Store.CMS.OriginatorInfo + Crypto.Store.CMS.PEM + Crypto.Store.CMS.Signed + Crypto.Store.CMS.Type + Crypto.Store.CMS.Util + Crypto.Store.Error + Crypto.Store.KeyWrap.AES + Crypto.Store.KeyWrap.RC2 + Crypto.Store.KeyWrap.TripleDES + Crypto.Store.PEM + Crypto.Store.PKCS12 + Crypto.Store.PKCS5 + Crypto.Store.PKCS5.PBES1 + Crypto.Store.PKCS8 + Crypto.Store.PKCS8.EC + Crypto.Store.Util + Crypto.Store.X509 + other-modules: + Paths_hs_hath + autogen-modules: + Paths_hs_hath + hs-source-dirs: + vendor/cryptostore-0.3.0.1/src + ghc-options: -Wall -Wcompat -Widentities -Wincomplete-record-updates -Wincomplete-uni-patterns -Wmissing-export-lists -Wmissing-home-modules -Wpartial-fields -Wredundant-constraints -w + build-depends: + aeson >=2.1 && <2.3 + , asn1-encoding ==0.9.* + , asn1-types >=0.3.1 && <0.4 + , async ==2.2.* + , base >=4.7 && <4.22 + , base64 ==1.0.* + , basement >=0.0.16 && <0.1 + , bytestring >=0.11 && <0.13 + , co-log >=0.6 && <0.8 + , co-log-core ==0.3.* + , co-log-polysemy >=0.0.1.5 && <0.0.2 + , conduit ==1.3.* + , containers >=0.6 && <0.8 + , crypton ==1.0.* + , crypton-connection ==0.4.* + , crypton-x509 ==1.7.* + , crypton-x509-store ==1.6.* + , crypton-x509-validation ==1.6.* + , dhall ==1.42.* + , directory ==1.3.* + , exceptions ==0.10.* + , extra >=1.7 && <1.9 + , filepath >=1.4 && <1.6 + , hex-text ==0.1.* + , hourglass ==0.2.* + , http-client ==0.7.* + , http-client-tls ==0.3.* + , http-conduit ==2.3.* + , http-media ==0.8.* + , http-types ==0.12.* + , lrucache ==1.2.* + , mason >=0.2.6 && <0.3 + , memory ==0.18.* + , monad-control ==1.0.* + , monad-loops ==0.4.* + , mtl ==2.3.* + , network >=3.1 && <3.3 + , optparse-applicative ==0.18.* + , pem >=0.1 && <0.3 + , polysemy ==1.9.* + , process ==1.6.* + , relude ==1.2.* + , resourcet ==1.3.* + , safe-exceptions ==0.1.* + , servant >=0.20.2 && <0.21 + , servant-client >=0.20.2 && <0.21 + , servant-client-core >=0.20.2 && <0.21 + , servant-server >=0.20.2 && <0.21 + , sqlite-simple ==0.4.* + , stm ==2.5.* + , string-interpolate ==0.3.* + , suspend ==0.2.* + , template-haskell >=2.19 && <2.24 + , text >=2.0 && <2.2 + , time >=1.12 && <1.15 + , timers ==0.2.* + , tls >=2.0 && <2.4 + , tls-session-manager >=0.0.5 && <0.1 + , transformers ==0.6.* + , unix ==2.8.* + , unliftio ==0.2.* + , unordered-containers ==0.2.* + , wai ==3.2.* + , wai-extra ==3.1.* + , warp ==3.4.* + default-language: GHC2021 + library hs-hath-minio exposed-modules: Lib.Prelude @@ -202,7 +309,6 @@ , crypton-connection ==0.* , crypton-x509 ==1.7.* , crypton-x509-store ==1.6.* - , cryptostore ==0.3.* , data-default-class ==0.* , dhall ==1.42.* , digest ==0.* @@ -288,7 +394,6 @@ , crypton-connection ==0.4.* , crypton-x509 ==1.7.* , crypton-x509-store ==1.6.* - , cryptostore ==0.3.* , dhall ==1.42.* , directory ==1.3.* , exceptions ==0.10.* @@ -392,7 +497,7 @@ build-depends: aeson >=2.1 && <2.3 , async ==2.2.* - , base ==4.21.* + , base >=4.18 && <4.22 , base64 ==1.0.* , bytestring >=0.11 && <0.13 , co-log >=0.6 && <0.8 @@ -406,14 +511,13 @@ , crypton-connection ==0.4.* , crypton-x509 ==1.7.* , crypton-x509-store ==1.6.* - , cryptostore ==0.3.* , dhall ==1.42.* , directory ==1.3.* , exceptions ==0.10.* , extra >=1.7 && <1.9 , filepath >=1.4 && <1.6 , ghc-compact ==0.1.* - , ghc-prim ==0.13.* + , ghc-prim >=0.10 && <0.14 , hex-text ==0.1.* , http-client ==0.7.* , http-client-tls ==0.3.* @@ -488,7 +592,6 @@ , crypton-connection ==0.4.* , crypton-x509 ==1.7.* , crypton-x509-store ==1.6.* - , cryptostore ==0.3.* , dhall ==1.42.* , directory ==1.3.* , exceptions ==0.10.* @@ -527,17 +630,95 @@ , text >=1 && <3 , time >=1.12 && <1.15 , timers ==0.2.* - , tls >=1 && <3 + , tls >=2.1.3 && <2.5 , tls-session-manager ==0.* , transformers ==0.6.* , unix ==2.8.* , unliftio ==0.2.* , unordered-containers ==0.2.* - , wai ==3.* + , wai ==3.2.* , wai-extra ==3.1.* - , warp ==3.* + , warp >=3.3.29 && <3.4.13 default-language: GHC2021 +executable hath-bench + main-is: Bench.hs + hs-source-dirs: + app + default-extensions: + NoImplicitPrelude + TypeApplications + OverloadedStrings + ScopedTypeVariables + QuasiQuotes + LambdaCase + ViewPatterns + MultiParamTypeClasses + GADTs + ghc-options: -Wall -Wcompat -Widentities -Wincomplete-record-updates -Wincomplete-uni-patterns -Wmissing-export-lists -Wmissing-home-modules -Wpartial-fields -Wredundant-constraints -threaded -rtsopts -with-rtsopts=-N -main-is Bench + build-depends: + aeson >=2.1 && <2.3 + , async ==2.2.* + , base >=4.7 && <4.22 + , base64 ==1.0.* + , bytestring >=0.11 && <0.13 + , co-log >=0.6 && <0.8 + , co-log-core ==0.3.* + , co-log-polysemy >=0.0.1.5 && <0.0.2 + , conduit ==1.3.* + , containers >=0.6 && <0.8 + , crypton ==1.0.* + , crypton-connection ==0.4.* + , crypton-x509 ==1.7.* + , crypton-x509-store ==1.6.* + , dhall ==1.42.* + , directory ==1.3.* + , exceptions ==0.10.* + , extra >=1.7 && <1.9 + , filepath >=1.4 && <1.6 + , hex-text ==0.1.* + , hs-hath + , http-client ==0.7.* + , http-client-tls ==0.3.* + , http-conduit ==2.3.* + , http-media ==0.8.* + , http-types ==0.12.* + , lrucache ==1.2.* + , mason >=0.2.6 && <0.3 + , memory ==0.18.* + , monad-control ==1.0.* + , monad-loops ==0.4.* + , mtl ==2.3.* + , network >=3.1 && <3.3 + , optparse-applicative ==0.18.* + , polysemy ==1.9.* + , process ==1.6.* + , relude ==1.2.* + , resourcet ==1.3.* + , safe-exceptions ==0.1.* + , servant >=0.20.2 && <0.21 + , servant-client >=0.20.2 && <0.21 + , servant-client-core >=0.20.2 && <0.21 + , servant-server >=0.20.2 && <0.21 + , sqlite-simple ==0.4.* + , stm ==2.5.* + , string-interpolate ==0.3.* + , suspend ==0.2.* + , template-haskell >=2.19 && <2.24 + , text >=2.0 && <2.2 + , time >=1.12 && <1.15 + , timers ==0.2.* + , tls >=2.0 && <2.4 + , tls-session-manager >=0.0.5 && <0.1 + , transformers ==0.6.* + , unix ==2.8.* + , unliftio ==0.2.* + , unordered-containers ==0.2.* + , wai ==3.2.* + , wai-extra ==3.1.* + , warp ==3.4.* + default-language: GHC2021 + executable hs-hath main-is: Main.hs hs-source-dirs: @@ -568,7 +749,6 @@ , crypton-connection ==0.4.* , crypton-x509 ==1.7.* , crypton-x509-store ==1.6.* - , cryptostore ==0.3.* , dhall ==1.42.* , directory ==1.3.* , exceptions ==0.10.* @@ -647,7 +827,6 @@ , crypton-connection ==0.4.* , crypton-x509 ==1.7.* , crypton-x509-store ==1.6.* - , cryptostore ==0.3.* , dhall ==1.42.* , directory , exceptions ==0.10.* @@ -703,7 +882,9 @@ main-is: Spec.hs other-modules: FilesystemSpec + FileURIRenderSpec Integration + InterfaceAPIOptimizationSpec MockRPC R2Spec Paths_hs_hath @@ -737,7 +918,6 @@ , crypton-connection ==0.4.* , crypton-x509 ==1.7.* , crypton-x509-store ==1.6.* - , cryptostore ==0.3.* , dhall ==1.42.* , directory , exceptions ==0.10.* @@ -745,6 +925,7 @@ , filepath , hex-text ==0.1.* , hs-hath + , hs-hath-polysemy-zoo , hspec ==2.11.* , http-client ==0.7.* , http-client-tls ==0.3.* diff --git a/src/HathNetwork/RPC.hs b/src/HathNetwork/RPC.hs --- a/src/HathNetwork/RPC.hs +++ b/src/HathNetwork/RPC.hs @@ -75,7 +75,7 @@ ClientStart -> log Info "Trying to start a session" >> startListening ClientStop -> log Info "Stopping a session" >> stopListening ClientLogin -> log Info "Logging in" >> login - CheckGalleryTask -> log Info "Checking for a gallery task" >> nextGalleryTask + CheckGalleryTask -> log Info "Checking for a gallery task" >> (rightToMaybe <$> nextGalleryTask) NotifyGalleryCompletion metadata -> log Info "Notifying server that a gallery task has been completed" >> completeGalleryTask metadata diff --git a/src/Interface/API.hs b/src/Interface/API.hs --- a/src/Interface/API.hs +++ b/src/Interface/API.hs @@ -10,8 +10,6 @@ API , api , ServerCommand(..) - , DynCT - , WithDynamicContentType(..) -- * EHentai RPC , EHentaiAPI(..) , ehRPC @@ -29,9 +27,10 @@ , completeGalleryTask , downloadGalleryFile , reportFailures - , fetchResource + , openResource , fetchBlacklist , runEHentaiAPIIO + , renderServerCommand ) where import Crypto.Store.PKCS12 ( readP12FileFromMemory @@ -43,6 +42,7 @@ import qualified Data.ByteString.Char8 as BSC import qualified Data.ByteString.Lazy as LBS +import qualified Data.List as List import Data.String.Interpolate ( i ) import qualified Data.Text as T import Data.Time.Clock.System ( SystemTime(systemSeconds), getSystemTime ) @@ -50,17 +50,23 @@ import qualified Mason.Builder as BD -import Network.HTTP.Client ( HttpException +import Network.HTTP.Client ( BodyReader + , HttpException , Manager , Request(responseTimeout, host, requestHeaders) - , Response(responseBody) + , Response(responseBody, responseHeaders, responseStatus) + , responseClose + , responseOpen , defaultManagerSettings , newManager , parseRequest , responseTimeoutMicro ) import Network.HTTP.Media ( (//), (/:) ) +import Network.HTTP.Client.TLS ( getGlobalManager ) import Network.HTTP.Simple ( httpLbs ) +import Network.HTTP.Types ( hContentLength ) +import qualified Network.HTTP.Types as HT import Polysemy import Polysemy.Error ( Error, errorToIOFinal, throw ) @@ -100,17 +106,19 @@ , HathSettings , RPCError(CertificateFailure) , RPCResponse(statusCode) - , emptyMetadata + , clientBuild , hentaiHeader , parseMetadata , parseRPCResponse , parseRPCResponse' , parseSettings + , renderFileURIBytes + , verifyRPCResponse ) import UnliftIO ( try ) -import Utils ( hash ) +import Utils ( firstJustM, hash ) -- | Available server commands data ServerCommand @@ -123,40 +131,22 @@ deriving ( Show, Eq ) -- Custom type conversion for ServerCommand -instance FromHttpApiData ServerCommand where - {-# INLINE parseUrlPiece #-} - parseUrlPiece = \case - "still_alive" -> Right StillAlive - "threaded_proxy_test" -> Right ThreadedProxyTest - "speed_test" -> Right SpeedTest - "refresh_settings" -> Right RefreshSettings - "start_downloader" -> Right StartDownloader - "refresh_certs" -> Right RefreshCerts - cmd -> Left $ "Invalid command: " <> cmd - -data DynCT - deriving ( Typeable ) - -instance MimeRender DynCT ByteString where - {-# INLINE mimeRender #-} - mimeRender _ = LBS.fromStrict - -instance Accept DynCT where - {-# INLINE contentType #-} - contentType _ = "" - -data WithDynamicContentType - = WithDynamicContentType - { contentType :: {-# UNPACK #-} !ByteString, content :: {-# UNPACK #-} !ByteString } +serverCommandNames :: [ ( Text, ServerCommand ) ] +serverCommandNames + = [ ( "still_alive", StillAlive ) + , ( "threaded_proxy_test", ThreadedProxyTest ) + , ( "speed_test", SpeedTest ) + , ( "refresh_settings", RefreshSettings ) + , ( "start_downloader", StartDownloader ) + , ( "refresh_certs", RefreshCerts ) + ] -instance MimeRender DynCT WithDynamicContentType where - {-# INLINE mimeRender #-} - mimeRender _ = LBS.fromStrict . content +instance FromHttpApiData ServerCommand where + parseUrlPiece cmd = maybeToRight ("Invalid command: " <> cmd) (List.lookup cmd serverCommandNames) -instance AllCTRender '[ DynCT ] WithDynamicContentType where - {-# INLINE handleAcceptH #-} - handleAcceptH _ _ (WithDynamicContentType ct content) - = Just ( LBS.fromStrict ct, LBS.fromStrict content ) +-- | Wire name of a command, needed to verify the servercmd signature. +renderServerCommand :: ServerCommand -> Text +renderServerCommand c = maybe "" fst (find ((== c) . snd) serverCommandNames) data SpeedTest deriving ( Typeable ) @@ -180,7 +170,7 @@ :> Capture "info" Text :> Capture "opts" Text :> Capture "filename" Text - :> Get '[DynCT] (Headers '[Header "Content-Length" Int] WithDynamicContentType) + :> RawM :<|> "metrics" :> Get '[PlainText] Text :<|> "servercmd" :> Capture "command" ServerCommand @@ -283,7 +273,7 @@ (runClientM (ehAPIM params { acttime = Just (show currentTime) - , clientbuild = Just (version cfg) + , clientbuild = Just clientBuild , actkey = Just $ makeKey params cfg currentTime , cid = Just (clientId cfg) }) @@ -295,8 +285,12 @@ pure servantReq { responseTimeout = responseTimeoutMicro rpcTimeout } }) {-# INLINE k #-} interpret (\case - EhRPC params -> k params "/15/rpc" - EhGallery params -> k params "/15/dl") m + -- get_cert is the raw PKCS#12 file, not a signed reply + EhRPC params + | act params == Just "get_cert" -> k params "/16/rpc" + | otherwise -> verifyRPCResponse (encodeUtf8 (key cfg)) <$> k params "/16/rpc" + -- the gallery list carries its own signature, checked by parseMetadata + EhGallery params -> k params "/16/dl") m where rpcTimeout :: Int rpcTimeout = 60 * 1000000 -- 60 seconds in microseconds @@ -382,12 +376,11 @@ parseSettings <$> parseRPCResponse' x {-# INLINE nextGalleryTask #-} -nextGalleryTask :: EHentaiAPI -@> Maybe GalleryMetadata +-- | Left says why there is nothing to download (NO_PENDING_DOWNLOADS, or a refused list). +nextGalleryTask :: [ EHentaiAPI, Reader ClientConfig ] >@> Either Text GalleryMetadata nextGalleryTask = do - m <- parseMetadata <$> ehGallery emptyRPCParams { act = Just "fetchqueue" } - if m == emptyMetadata - then return Nothing - else return $ Just m + cfg <- ask @ClientConfig + parseMetadata (encodeUtf8 (key cfg)) <$> ehGallery emptyRPCParams { act = Just "fetchqueue" } {-# INLINE completeGalleryTask #-} completeGalleryTask :: GalleryMetadata -> EHentaiAPI -@> () @@ -415,7 +408,7 @@ >>= \case Left _ -> pure Nothing Right bytes -> do - if hash bytes /= galleryFileHash file + if maybe False (/= hash bytes) (galleryFileHash file) then do modify @[ Text ] @@ -436,37 +429,44 @@ if null urls then operate (retries + 1) -- we did not get a valid URL, might be a network error - else maybe (operate (retries + 1)) (pure . Just) . asum - =<< traverse (download . BSC.unpack) urls + else do + -- `traverse` would execute every fallback download eagerly. + -- We need monadic short-circuiting to stop after the first success. + firstJustM (download . BSC.unpack) urls >>= maybe (operate (retries + 1)) (pure . Just) {-# INLINE reportFailures #-} reportFailures :: [ Text ] -> EHentaiAPI -@> () reportFailures reports = void $ ehRPC emptyRPCParams { act = Just "dlfails", add = Just $ T.intercalate ";" reports } -fetchResource :: FileURI - -> ( ByteString, ByteString ) - -> [ EHentaiAPI, Reader ClientConfig, Error RPCError, Embed IO ] >@> Maybe ByteString -fetchResource fileURI ( fileIndex, xres ) = do +-- | Ask the RPC server where a static-range file lives and open the first source +-- that answers 200 with the expected length, as the official ProxyFileDownloader +-- does. The caller streams the body and must close the response. +openResource :: FileURI + -> ( ByteString, ByteString ) + -> [ EHentaiAPI, Reader ClientConfig, Error RPCError, Embed IO ] >@> Maybe (Response BodyReader) +openResource fileURI ( fileIndex, xres ) = do urls <- parseRPCResponse' =<< ehRPC emptyRPCParams { act = Just "srfetch", add = Just [i|#{fileIndex};#{xres};#{fileURI}|] } cfg <- ask @ClientConfig - asum - <$> traverse - (download [i|#{clientId cfg}-#{hash @ByteString $ encodeUtf8 (key cfg) <> show fileURI}|] - . BSC.unpack) - urls + let token = [i|#{clientId cfg}-#{hash @ByteString $ encodeUtf8 (key cfg) <> renderFileURIBytes fileURI}|] + embed $ do + mgr <- getGlobalManager + firstJustM (open mgr token . BSC.unpack) urls where - download token url = case parseRequest url of + open mgr token url = case parseRequest url of Nothing -> pure Nothing - Just req -> do - bytes <- embed - $ LBS.toStrict . responseBody - <$> httpLbs @IO req { requestHeaders = ( "Hath-Request", token ) : hentaiHeader } - if hash bytes /= fileHash fileURI - then pure Nothing - else return $ Just bytes + Just req -> try @IO @HttpException + (responseOpen req { requestHeaders = ( "Hath-Request", token ) : hentaiHeader } mgr) + >>= \case + Right resp + | HT.statusCode (responseStatus resp) == 200 + , List.lookup hContentLength (responseHeaders resp) == Just (show (fileSize fileURI)) -> + pure (Just resp) + | otherwise -> responseClose resp $> Nothing + -- connection error or wrong answer: firstJustM tries the next source + Left _ -> pure Nothing fetchBlacklist :: Int -> [ EHentaiAPI, Error RPCError ] >@> [ Text ] fetchBlacklist deltaTime = do diff --git a/src/LegacyCiphers.hs b/src/LegacyCiphers.hs --- a/src/LegacyCiphers.hs +++ b/src/LegacyCiphers.hs @@ -139,34 +139,34 @@ aes128cbc BulkEncrypt key = let ctx = noFail (cipherInit key) :: AES128 - in + in (\iv input -> let output = cbcEncrypt ctx (makeIV_ iv) input - in + in ( output, takelast 16 output )) aes128cbc BulkDecrypt key = let ctx = noFail (cipherInit key) :: AES128 - in + in (\iv input -> let output = cbcDecrypt ctx (makeIV_ iv) input - in + in ( output, takelast 16 input )) aes256cbc :: BulkDirection -> BulkKey -> BulkBlock aes256cbc BulkEncrypt key = let ctx = noFail (cipherInit key) :: AES256 - in + in (\iv input -> let output = cbcEncrypt ctx (makeIV_ iv) input - in + in ( output, takelast 16 output )) aes256cbc BulkDecrypt key = let ctx = noFail (cipherInit key) :: AES256 - in + in (\iv input -> let output = cbcDecrypt ctx (makeIV_ iv) input - in + in ( output, takelast 16 input )) diff --git a/src/Middleware.hs b/src/Middleware.hs --- a/src/Middleware.hs +++ b/src/Middleware.hs @@ -9,10 +9,15 @@ , IP(..) , IPRecord(..) , timeWindow + , checkRateLimit + , isLocalIP + , parseIPv4 ) where import Control.Exception ( finally ) +import Data.Bits ( shiftR, (.&.) ) + import qualified Data.ByteString as BS import qualified Data.HashMap.Strict as HashMap @@ -21,7 +26,13 @@ import qualified Metrics.Gauge as Gauge import Network.HTTP.Types ( hAccept ) -import Network.Socket ( HostAddress, HostAddress6, SockAddr(..) ) +import Network.Socket ( HostAddress + , HostAddress6 + , SockAddr(..) + , hostAddress6ToTuple + , hostAddressToTuple + , tupleToHostAddress + ) import Network.Wai ( Middleware , mapResponseHeaders , rawPathInfo @@ -39,18 +50,23 @@ import Utils ( tooManyRequestsResponse ) -timeWindow :: Double -timeWindow = 10 * 60 -- 10 minutes in seconds +-- | Mirrors the official client's flood control (HTTPServer.FloodControlEntry): +-- every request adds one hit, hits drain at one per second, and more than +-- 'floodLimit' hits blocks the IP for 'blockSeconds'. +floodLimit :: Int +floodLimit = 10 -maxRequests :: Int8 -maxRequests = 5 +blockSeconds :: Double +blockSeconds = 60 --- Data types for tracking requests --- Note: Lists cannot be unpacked, so we use strict spine with explicit strictness +-- | Entries idle for this long carry no state worth keeping (any block has expired). +timeWindow :: Double +timeWindow = 60 + data IPRecord - = IPRecord { earlistRequestTime :: !Double -- ^ Time of the first request - , requestTimes :: !Int8 -- ^ Number of requests in the time window - , bannedUntil :: !Double -- ^ When the ban expires + = IPRecord { hitCount :: !Int -- ^ Hits left after draining + , lastHit :: !Double -- ^ Monotonic time of the last counted hit + , bannedUntil :: !Double -- ^ When the block expires } deriving ( Show ) @@ -73,64 +89,79 @@ globalIPMap :: IORef IPMap globalIPMap = unsafePerformIO $ newIORef HashMap.empty -{-# NOINLINE globalIPMapCounter #-} -globalIPMapCounter :: IORef Int -globalIPMapCounter = unsafePerformIO $ newIORef 0 - --- Create a new rate limiting middleware -rateLimitMiddleware :: Middleware -rateLimitMiddleware app req k +-- | Flood control for /h/. Local addresses and those the predicate exempts +-- (the RPC servers, as in the official client) are never limited. +rateLimitMiddleware :: (IP -> IO Bool) -> Middleware +rateLimitMiddleware exempt app req k | "/h/" `BS.isPrefixOf` rawPathInfo req = do now <- getMonotonicTime let sockAddr = remoteHost req case getIP sockAddr of Nothing -> app req k -- If we can't get IP, just allow the request - Just ip -> do - allowed <- checkRateLimit ip now - if allowed - then app req k - else k tooManyRequestsResponse + Just ip + | isLocalIP ip -> app req k + | otherwise -> do + skip <- exempt ip + allowed <- if skip then pure True else checkRateLimit ip now + if allowed + then app req k + else k tooManyRequestsResponse | otherwise = app req k -- Skip rate limiting for non-resource paths --- Check if request is allowed and update rate limit state --- --- return False if the request should be blocked -checkRateLimit :: IP -> Double -> IO Bool -checkRateLimit ip now = do - cnt <- readIORef globalIPMapCounter - if cnt > 100 - then do - writeIORef globalIPMapCounter 0 - writeIORef globalIPMap HashMap.empty - pure True - else do - writeIORef globalIPMapCounter (cnt + 1) - m <- readIORef globalIPMap - maybe firstRequest existingRequest (HashMap.lookup ip m) +-- | "a.b.c.d" or "::ffff:a.b.c.d", the form rpc_server_ip uses. +-- ponytail: IPv4 only; the RPC servers have no IPv6 addresses. +parseIPv4 :: ByteString -> Maybe IP +parseIPv4 s = case traverse (readMaybe @Int . decodeUtf8) (BS.split 46 v4) of + Just [ a, b, c, d ] + | all (\x -> x >= 0 && x <= 255) [ a, b, c, d ] -> Just + (IPv4 (tupleToHostAddress ( fromIntegral a, fromIntegral b, fromIntegral c, fromIntegral d ))) + _ -> Nothing where - -- Handle first request from an IP by creating new record - firstRequest = do - modifyIORef' globalIPMap $ HashMap.insert ip (IPRecord now 1 0) - return True + v4 = fromMaybe s (BS.stripPrefix "::ffff:" s) - -- Check if IP is banned, otherwise check request count - existingRequest record - | bannedUntil record > now = return False - | otherwise = checkAndUpdateRequests record +-- | Loopback, private and link-local addresses, which the official client +-- exempts from flood control (HTTPServer.localNetworkPattern). +isLocalIP :: IP -> Bool +isLocalIP (IPv4 ha) = localV4 (hostAddressToTuple ha) +isLocalIP (IPv6 ha) = case hostAddress6ToTuple ha of + ( 0, 0, 0, 0, 0, 0, 0, 1 ) -> True + -- IPv4-mapped (::ffff:a.b.c.d), as seen on dual-stack sockets + ( 0, 0, 0, 0, 0, 0xffff, hi, lo ) -> localV4 ( byte hi 8, byte hi 0, byte lo 8, byte lo 0 ) + ( w, _, _, _, _, _, _, _ ) -> (w .&. 0xfe00) == 0xfc00 -- fc00::/7 + where + byte w n = fromIntegral (w `shiftR` n) - -- Check if requests are within rate limit window and update accordingly - checkAndUpdateRequests record - = if now - earlistRequestTime record <= timeWindow - then if requestTimes record > maxRequests - then do - -- Ban the IP for 5 minutes - modifyIORef' globalIPMap $ HashMap.insert ip (record { bannedUntil = now + 5 * 60 }) - return False - else do - modifyIORef' globalIPMap - $ HashMap.insert ip (record { requestTimes = requestTimes record + 1 }) - return True - else return True +localV4 :: ( Word8, Word8, Word8, Word8 ) -> Bool +localV4 = \case + ( 127, _, _, _ ) -> True + ( 10, _, _, _ ) -> True + ( 192, 168, _, _ ) -> True + ( 172, b, _, _ ) -> b >= 16 && b <= 31 + ( 169, 254, _, _ ) -> True + _ -> False + +-- | Count a hit for the IP; False if the request should be blocked. +checkRateLimit :: IP -> Double -> IO Bool +checkRateLimit ip now = atomicModifyIORef' globalIPMap $ \m0 -> + let + -- ponytail: O(n) prune once the map is large; fine for the few hundred IPs a client sees per minute + m = if HashMap.size m0 > 1000 + then HashMap.filter ((> now - timeWindow) . lastHit) m0 + else m0 + in + case HashMap.lookup ip m of + Just r + | bannedUntil r > now -> ( m, False ) -- blocked requests are not counted, as in the official client + old -> let + r0 = fromMaybe (IPRecord 0 now 0) old + hits = max 0 (hitCount r0 - floor (now - lastHit r0)) + 1 + blocked = hits > floodLimit + r = IPRecord { hitCount = hits + , lastHit = now + , bannedUntil = if blocked then now + blockSeconds else bannedUntil r0 + } + in + ( HashMap.insert ip r m, not blocked ) type KeystampMap = HashMap ByteString ( Int, Double ) diff --git a/src/Server.hs b/src/Server.hs --- a/src/Server.hs +++ b/src/Server.hs @@ -7,6 +7,7 @@ ( startServer , ServerAction(..) , makeApplication + , mkServerLoopContext , CacheRunner(..) , IPMap , KeystampMap @@ -26,13 +27,13 @@ import qualified Data.ByteString as BS import qualified Data.ByteString.Char8 as BSC +import Data.ByteString.Builder ( byteString ) import qualified Data.ByteString.Lazy as LBS import qualified Data.ByteString.Lazy.Char8 as LBS8 import qualified Data.HashMap.Strict as HashMap import qualified Data.Map.Strict as Map import Data.String.Interpolate ( i ) import qualified Data.Text as Text -import Data.Time.Clock.POSIX ( POSIXTime, getPOSIXTime ) import Data.Time.Clock.System ( SystemTime(systemSeconds), getSystemTime ) import Data.X509 ( CertificateChain, PrivKey ) @@ -49,11 +50,11 @@ import Interface.API ( API , EHentaiAPI , ServerCommand(..) - , WithDynamicContentType(WithDynamicContentType) , api , downloadGalleryFile , fetchBlacklist , nextGalleryTask + , renderServerCommand , runEHentaiAPI , runEHentaiAPIIO , startListening @@ -66,19 +67,27 @@ import Middleware -import Network.HTTP.Client ( brReadSome +import qualified Network.HTTP.Client as HC +import Network.HTTP.Client ( Manager + , brRead , newManager , parseRequest , withResponse ) import Network.HTTP.Client.TLS ( tlsManagerSettings ) -import Network.HTTP.Simple ( getResponseBody ) -import Network.HTTP.Types ( status200, status404 ) +import Network.HTTP.Simple ( getResponseBody, getResponseStatusCode ) +import qualified Network.HTTP.Types as HT +import Network.HTTP.Types ( hContentLength + , hContentType + , status200 + , status404 + ) import Network.TLS ( Credentials(Credentials), Version(..) ) import Network.TLS.Extra ( ciphersuite_all ) import Network.Wai ( Middleware , Request(requestMethod) , responseLBS + , responseStream ) import Network.Wai.Handler.Warp ( defaultSettings, setPort ) import Network.Wai.Handler.WarpTLS ( OnInsecure(AllowInsecure) @@ -142,7 +151,7 @@ import Types ( CacheBackend(..) , ClientConfig , ClientConfig(..) - , FileURI(fileExt) + , FileURI(fileExt, fileSize) , GalleryFile(..) , GalleryMetadata(..) , HathSettings(..) @@ -158,6 +167,7 @@ , race , readTChan , replicateConcurrently + , timeout , withAsync , writeTChan ) @@ -172,6 +182,15 @@ maxTimeDrift :: Int64 maxTimeDrift = 600 +-- | Official Settings.MAX_KEY_TIME_DRIFT, used for servercmd. +maxCommandTimeDrift :: Int64 +maxCommandTimeDrift = 300 + +-- | Whole-download cap for one threaded_proxy_test request. +-- ponytail: one total timeout; the official client uses 10 s connect + 60 s per read. +testTimeoutMicros :: Int +testTimeoutMicros = 60 * 1000000 + data ServerAction = Reload | Cert | GracefulShutdown data GalleryTask = GalleryTask @@ -220,6 +239,34 @@ -- ^ Cache backend interpreter (SQLite, R2, or Filesystem) } +mkServerLoopContext + :: ClientConfig + -> TVar HathSettings + -> TChan ServerAction + -> TChan GalleryTask + -> Bool + -> Bool + -> CacheRunner + -> StatsEnv + -> ServerLoopContext +mkServerLoopContext cfg settings action gallery disableRL trustProxy cacheRunner statsEnv + = ServerLoopContext + { loopConfig = cfg + , loopChannels = ServerChannels + { serverActionChan = action + , galleryTaskChan = gallery + } + , loopState = ServerState + { serverSettings = settings + , serverStatsEnv = statsEnv + } + , loopFlags = ServerFlags + { disableRateLimit = disableRL + , trustProxyHeaders = trustProxy + } + , loopCacheRunner = cacheRunner + } + -- Server implementation server :: Members '[ Embed IO @@ -257,7 +304,12 @@ resourceHandler (encodeUtf8 -> fileid) (parseURLParams . encodeUtf8 -> opts) - (encodeUtf8 -> filename) = do + (encodeUtf8 -> filename) + req + respond + -- answer HEAD like any other path instead of opening an upstream body nobody reads + | requestMethod req == "HEAD" = embed $ respond $ responseLBS status200 hentaiHeader "" + | otherwise = do currentTime <- embed getSystemTime cfg <- ask @ClientConfig when (abs (timestamp - systemSeconds currentTime) > maxTimeDrift) @@ -267,24 +319,33 @@ LocateURI { locateURIFilename = filename, locateURI = uri, locateURIOptions = opts } >>= \case Nothing -> throw err404 - Just (Redirect url) -> throw + Just (Stored (Redirect url)) -> throw $ err302 { errHeaders = [ ( "Location", url ), ( "Cache-Control", "public, max-age=600" ) ] } - Just (Record (fileRecordBytes -> bs)) -> do + Just (Stored (Record (fileRecordBytes -> bs))) -> do incServed addUpload (BS.length bs) - return - $ addHeader @"Content-Length" (BS.length bs) - $ WithDynamicContentType mimeType bs + embed $ respond $ responseLBS status200 (fileHeaders (BS.length bs)) (LBS.fromStrict bs) + -- cache miss: relay upstream bytes as they arrive, like the official proxy mode + Just (Streamed relay) -> do + incServed + addUpload (fileSize uri) + embed + $ respond + $ responseStream status200 (fileHeaders (fileSize uri)) + $ \write flush -> relay (\chunk -> write (byteString chunk) >> flush) where + fileHeaders :: Int -> [ HT.Header ] + fileHeaders len = [ ( hContentType, mimeType ), ( hContentLength, show len ) ] + mimeType = case fileExt uri of "jpg" -> "image/jpeg" "png" -> "image/png" "gif" -> "image/gif" "mp4" -> "video/mp4" "wbm" -> "video/webm" - "wbp" -> "video/webp" - "avf" -> "video/avif" + "wbp" -> "image/webp" + "avf" -> "image/avif" "jxl" -> "image/jxl" _ -> "application/octet-stream" @@ -295,7 +356,7 @@ rs = fromMaybe "" (Map.lookup "keystamp" opts) ( t, rk ) = BSC.span (/= '-') rs in - ( fromIntegral $ maybe 0 fst $ BSC.readInteger t, BS.tail rk ) + ( fromIntegral $ maybe 0 fst $ BSC.readInteger t, BS.drop 1 rk ) {-# INLINE challange #-} -- [i|#{timestamp}-#{fileid}-#{key cfg}-hotlinkthis|] @@ -311,8 +372,9 @@ <> BD.textUtf8 (key cfg) <> "-hotlinkthis") - serverCmdHandler command (parseURLParams . encodeUtf8 -> additional) _time _key - = case command of + serverCmdHandler command rawAdditional commandTime commandKey + = checkCommandKey + >> case command of StillAlive -> plainText "I feel FANTASTIC and I'm still alive" ThreadedProxyTest -> let args = do @@ -351,12 +413,10 @@ Nothing -> throw err403 Just req -> do results <- embed $ replicateConcurrently testCount $ runTest testSize req mgr - let ( failed :: Int, millis :: Int64 ) - = foldl' (\( f, t ) r -> case r of - Nothing -> ( f + 1, t ) - Just v -> ( f, t + v )) ( 0, 0 ) results + let succeeded = catMaybes results addUpload (testCount * fromIntegral testSize) - plainText [i|OK:#{failed}-#{millis}|] + -- official format: OK:- + plainText [i|OK:#{length succeeded}-#{sum succeeded}|] SpeedTest -> let testSize = maybe @@ -388,17 +448,45 @@ atomically $ writeTChan chan Cert plainText "" where + additional = parseURLParams (encodeUtf8 rawAdditional) + + -- same check as the official client: only the RPC server knows our key + checkCommandKey = do + currentTime <- embed getSystemTime + cfg <- ask @ClientConfig + let expected + = hash @Text + $ encodeUtf8 @Text + [i|hentai@home-servercmd-#{renderServerCommand command}-#{rawAdditional}-#{clientId cfg}-#{commandTime}-#{key cfg}|] + when (abs (fromIntegral commandTime - systemSeconds currentTime) > maxCommandTimeDrift + || commandKey /= expected) + $ throw err403 + + -- A test succeeds only if the whole body arrives with status 200 within the + -- timeout (official: FileDownloader + getContentLength() >= testsize). + runTest :: Int64 -> HC.Request -> Manager -> IO (Maybe Int64) runTest testSize req mgr = do - start <- getPOSIXTime - let phi 0 _ = do - end <- getPOSIXTime - return $ truncate $ realToFrac @POSIXTime @Double $ 1000 * (end - start) - phi n br = do - s <- LBS.length <$> brReadSome br 1024 - phi (max 0 (n - s)) br - try @SomeException (withResponse req mgr (phi testSize . getResponseBody)) >>= \case - Left _ -> return Nothing - Right v -> return $ Just v + start <- getMonotonicTime + ok <- timeout testTimeoutMicros + $ try @SomeException + $ withResponse req mgr + $ \resp -> let + drain n + | n <= 0 = pure True + | otherwise = do + chunk <- brRead (getResponseBody resp) + -- an empty chunk is EOF: the body was shorter than promised + if BS.null chunk + then pure False + else drain (n - fromIntegral (BS.length chunk)) + in + if getResponseStatusCode resp == 200 + then drain testSize + else pure False + end <- getMonotonicTime + pure $ case ok of + Just (Right True) -> Just (truncate ((end - start) * 1000)) + _ -> Nothing {-# INLINE readInt #-} {-# SPECIALISE readInt :: ByteString -> Maybe Int #-} @@ -506,8 +594,11 @@ applyRateLimit = if disableRL then id - else rateLimitMiddleware + else rateLimitMiddleware isRpcServer + -- dual-stack sockets would show RPC servers as ::ffff:a.b.c.d; Warp listens on IPv4 here + isRpcServer ip = (\st -> Just ip `elem` (parseIPv4 <$> rpcServerIPs st)) <$> readTVarIO settings + finalMiddleware :: Middleware finalMiddleware = applyRealIp . applyRateLimit @@ -595,11 +686,13 @@ -- This function repeatedly calls 'stopListening' until the RPC server -- acknowledges the shutdown request, then exits the process cleanly. gracefulShutdown :: ClientConfig -> IO a -gracefulShutdown cfg = go +gracefulShutdown cfg = go (3 :: Int) where - go = runEHentaiAPIIO cfg stopListening >>= \case - Right _ -> exitSuccess - Left _ -> go + -- The official client sends client_stop once. Never spin when the RPC server is unreachable. + go n = runEHentaiAPIIO cfg stopListening >>= \case + Left _ + | n > 1 -> threadDelay 1000000 >> go (n - 1) + _ -> exitSuccess -- | Refresh TLS certificates with exponential backoff retry logic. -- @@ -700,8 +793,9 @@ metadata <- nextGalleryTask incDlTask case metadata of - Nothing -> log Info "No gallery task available" - Just md -> do + Left "NO_PENDING_DOWNLOADS" -> log Info "No gallery task available" + Left reason -> log Warning [i|Gallery download refused: #{reason}|] + Right md -> do log Info [i|Starting download for gallery #{galleryID md}|] complete <- mapM (verifyAndDownload md) (galleryFileList md) when (and complete) (notifyGalleryCompletion md) @@ -715,7 +809,8 @@ if existing then BS.readFile filePath else return BS.empty - if galleryFileHash f == hash existingBytes + -- a file generated on demand has no known hash; like the official client, keep it + if not (BS.null existingBytes) && maybe True (== hash existingBytes) (galleryFileHash f) then log Info [i|#{galleryFileName f}.#{galleryFileExt f} already verified, skipping|] >> pure True else downloadGalleryFile md f >>= \case diff --git a/src/Stats.hs b/src/Stats.hs --- a/src/Stats.hs +++ b/src/Stats.hs @@ -95,7 +95,6 @@ dlBytesCounter errorCounter latencyGauge - data Stats m a where AddUpload :: Int -> Stats m () AddDownload :: Int -> Stats m () diff --git a/src/Storage/Database.hs b/src/Storage/Database.hs --- a/src/Storage/Database.hs +++ b/src/Storage/Database.hs @@ -16,7 +16,12 @@ import Relude hiding ( Reader, State, ask, evalState, get, modify, put ) -import Types ( FileRecord(..), FileURI, StorageResult(..) ) +import Types ( FileRecord(..) + , FileURI + , StorageResult(..) + , renderFileURIBytes + , renderFileURIText + ) {-# INLINE initializeDB #-} -- | Initialize database with required schema @@ -42,7 +47,7 @@ :: Members '[ Embed IO ] r => Connection -> KVStore FileURI StorageResult : r @> a -> r @> a runCache conn = interpret $ \case LookupKV uri -> do - let fid = show @Text uri + let fid = renderFileURIText uri embed $ execute conn "UPDATE files SET lru_counter = lru_counter + 1 WHERE file_id = ?" (Only fid) results <- embed @@ -67,7 +72,7 @@ record UpdateKV uri Nothing -> let - fid = show @ByteString uri + fid = renderFileURIBytes uri in embed $ execute conn "DELETE FROM files WHERE file_id = ?" (Only fid) diff --git a/src/Storage/Filesystem.hs b/src/Storage/Filesystem.hs --- a/src/Storage/Filesystem.hs +++ b/src/Storage/Filesystem.hs @@ -28,7 +28,12 @@ import System.Directory ( createDirectoryIfMissing, doesFileExist, removeFile ) import System.FilePath ( (), takeDirectory, takeFileName ) -import Types ( ClientConfig, FileURI(..), RPCError(..), StorageResult(..) ) +import Types ( ClientConfig + , FileURI(..) + , RPCError(..) + , StorageResult(..) + , renderFileURIText + ) import qualified Types as Ty import Utils ( log ) @@ -39,7 +44,7 @@ fileURIToPath root uri = root s4 fileId where -- Sanitize to prevent directory traversal - fileId = takeFileName (show uri) + fileId = takeFileName (toString $ renderFileURIText uri) s4 = T.unpack $ T.take 4 (T.pack fileId) diff --git a/src/Storage/Locate.hs b/src/Storage/Locate.hs --- a/src/Storage/Locate.hs +++ b/src/Storage/Locate.hs @@ -1,20 +1,33 @@ {-# LANGUAGE DataKinds #-} {-# LANGUAGE TemplateHaskell #-} -module Storage.Locate ( LocateURI(..), Locate(..), locateResource, runLocate ) where +module Storage.Locate + ( LocateURI(..) + , Locate(..) + , Located(..) + , locateResource + , runLocate + , streamUpstream + ) where import Colog ( Message, Severity(Info, Warning) ) import Colog.Polysemy ( Log ) +import Control.Concurrent ( forkIO ) +import Control.Exception ( finally ) + import qualified Data.ByteString as BS import qualified Data.ByteString.Short as SBS import qualified Data.Map as Map import Data.String.Interpolate ( i ) -import Interface.API ( EHentaiAPI, fetchResource ) +import Interface.API ( EHentaiAPI, openResource ) +import qualified Network.HTTP.Client as HC + import Polysemy -import Polysemy.Error ( Error ) +import Polysemy.Error ( Error, catch ) +import Polysemy.Final ( withWeavingToFinal ) import Polysemy.KVStore ( KVStore, lookupKV, updateKV ) import Polysemy.Operators import Polysemy.Reader ( Reader ) @@ -27,7 +40,7 @@ import Types -import Utils ( log ) +import Utils ( hash, log ) data LocateURI = LocateURI { locateURIFilename :: !ByteString @@ -35,9 +48,15 @@ , locateURIOptions :: !(Map ByteString ByteString) } +data Located + = Stored !StorageResult + -- | Cache miss being relayed from upstream: call it with a chunk writer. + -- It closes the upstream connection and caches the file once it is complete. + | Streamed !((ByteString -> IO ()) -> IO ()) + data Locate m a where -- | Fetch a specific resource from the server - LocateResource :: LocateURI -> Locate m (Maybe StorageResult) + LocateResource :: LocateURI -> Locate m (Maybe Located) makeSem ''Locate @@ -49,6 +68,7 @@ , Reader ClientConfig , EHentaiAPI , Embed IO + , Final IO , Stats , Log Message ] @@ -59,33 +79,36 @@ LocateResource uri -> do settings <- getSettings let fileURI = locateURI uri - s4 = SBS.take 4 . show $ fileURI + s4 = SBS.toShort $ BS.take 4 $ renderFileURIBytes fileURI + record content + = Record + FileRecord + { fileRecordLRUCounter = 1 + , fileRecordS4 = decodeUtf8 s4 + , fileRecordFileId = renderFileURIText fileURI + , fileRecordFileName = Just $ decodeUtf8 $ locateURIFilename uri + , fileRecordBytes = content + } -- if HashSet.member s4 (staticRanges settings) <-- we have a bug in setting parsing. wait for fix if True then lookupKV fileURI >>= \case - Just x@Redirect {} -> return $ Just x - Just x@Record {} -> return $ Just x + Just x -> return $ Just $ Stored x Nothing -> case ( Map.lookup "fileindex" (locateURIOptions uri) , Map.lookup "xres" (locateURIOptions uri) ) of ( Just fileIndex, Just xres ) -> log Info [i|Fetching resource: #{fileURI}|] - >> fetchResource fileURI ( fileIndex, xres ) + >> openResource fileURI ( fileIndex, xres ) >>= \case - Just content -> do + Just resp -> do incFetched - addDownload (BS.length content) - let record - = Record - FileRecord - { fileRecordLRUCounter = 1 - , fileRecordS4 = decodeUtf8 s4 - , fileRecordFileId = show fileURI - , fileRecordFileName = Just $ decodeUtf8 $ locateURIFilename uri - , fileRecordBytes = content - } - updateKV fileURI $ Just record - pure $ Just record - Nothing -> do + -- runs after the handler has returned, once the body has been relayed + cacheIt <- deferToIO $ \content -> do + addDownload (BS.length content) + -- a failed cache write only costs us the next hit + catch @RPCError (updateKV fileURI $ Just $ record content) + $ \e -> log Warning [i|Cache write failed for #{fileURI}: #{e}|] + pure $ Just $ Streamed $ streamUpstream fileURI resp (void . forkIO . cacheIt) + Nothing -> do log Warning [i|Failed to fetch resource, sorry for you|] pure Nothing _ -> do @@ -94,3 +117,25 @@ else do log Info [i|Resource not in our ranges, rejecting: #{s4}|] pure Nothing + +-- | Relay an upstream body chunk by chunk. If it arrives complete and its SHA-1 +-- matches, hand the bytes to the continuation (the cache write). The upstream +-- connection is closed in every case, including the client going away. +streamUpstream :: FileURI + -> HC.Response HC.BodyReader + -> (ByteString -> IO ()) + -> (ByteString -> IO ()) + -> IO () +streamUpstream fileURI resp whenComplete write = go [] `finally` HC.responseClose resp + where + go acc = HC.brRead (HC.responseBody resp) >>= \chunk -> if BS.null chunk + then let + content = BS.concat (reverse acc) + in + when (BS.length content == fileSize fileURI && hash content == fileHash fileURI) + $ whenComplete content + else write chunk >> go (chunk : acc) + +-- | Lower a Sem action into IO so it can run after the request handler has returned. +deferToIO :: Member (Final IO) r => (x -> Sem r ()) -> Sem r (x -> IO ()) +deferToIO k = withWeavingToFinal $ \s lower _ -> pure ((\x -> void (lower (k x <$ s))) <$ s) diff --git a/src/Storage/R2.hs b/src/Storage/R2.hs --- a/src/Storage/R2.hs +++ b/src/Storage/R2.hs @@ -38,23 +38,28 @@ import Polysemy.Error ( Error, throw ) import Polysemy.KVStore ( KVStore(..) ) import Polysemy.Operators -import Polysemy.Reader ( Reader, asks ) import Relude hiding ( Reader, ask, asks ) import Storage.Database ( FileRecord(..) ) -import Types ( ClientConfig - , FileURI(..) +import Types ( FileURI(..) , RPCError(..) , StorageResult(..) + , renderFileURIText ) import qualified Types as Ty import Utils ( log ) -- | R2 connection configuration built at runtime -data R2Connection = R2Connection { r2MinioConn :: !Minio.MinioConn, r2ConnBucket :: !Text } +data R2Connection + = R2Connection + { r2MinioConn :: !Minio.MinioConn + , r2ConnBucket :: !Text + , r2UrlCache :: !(AtomicLRU FileURI ( Double, ByteString )) + -- ^ Presigned links. Lives here because runCacheR2 runs once per request. + } -- | Build R2 connection from config + environment variables -- Reads R2_ACCESS_KEY and R2_SECRET_KEY from environment @@ -78,29 +83,30 @@ minioConn <- liftIO $ do manager <- newManager tlsManagerSettings Minio.mkMinioConn (setCreds creds $ setRegion "auto" baseConnInfo) manager - pure $ Right R2Connection { r2MinioConn = minioConn, r2ConnBucket = Ty.r2Bucket cfg } + -- ponytail: fixed capacity; links expire after 10 min, so it only has to hold ~10 min of distinct hits + urlCache <- newAtomicLRU (Just 100000) + pure + $ Right + R2Connection + { r2MinioConn = minioConn, r2ConnBucket = Ty.r2Bucket cfg, r2UrlCache = urlCache } -- | Generate R2 object key from FileURI -- Format: s4/hash-size-xres-yres-ext fileURIToKey :: FileURI -> Text fileURIToKey uri = s4 <> "/" <> fileId where - fileId = show uri + fileId = renderFileURIText uri s4 = T.take 4 fileId -- | Run the cache with Cloudflare R2 -- Read operations silently degrade to Nothing on failure (with warning log) -- Write/delete operations propagate errors -runCacheR2 :: Members '[ Embed IO, Log Message, Reader ClientConfig, Error RPCError ] r +runCacheR2 :: Members '[ Embed IO, Log Message, Error RPCError ] r => R2Connection -> KVStore FileURI StorageResult : r @> a -> r @> a -runCacheR2 conn m = do - lruSize <- asks Ty.lruCacheSize - let entries = Just (fromIntegral (lruSize `div` 300)) -- Approximate number of entries based on avg url link size ~300 bytes - memCache <- embed $ newAtomicLRU entries - interpret (phi memCache) m +runCacheR2 conn = interpret (phi (r2UrlCache conn)) where bucket = r2ConnBucket conn diff --git a/src/Types.hs b/src/Types.hs --- a/src/Types.hs +++ b/src/Types.hs @@ -18,28 +18,34 @@ , StorageResult(..) -- * Globals , hentaiHeader + , clientBuild -- * Default values , defaultHathSettings , defaultClientConfig , emptyFileURI - , emptyMetadata -- * Parsing , parseSettings , parseMetadata , readClientConfig , parseFileURI + , renderFileURIBytes + , renderFileURIText , parseRPCResponse , parseRPCResponse' + , verifyRPCResponse , reconstructRecord -- * Selectors , getPayload ) where import qualified Data.ByteString.Char8 as BS +import Data.Char ( isAsciiLower, isAsciiUpper, isDigit ) import qualified Data.ByteString.Short as SBS import qualified Data.HashSet as HashSet import qualified Data.Text as T +import qualified Mason.Builder as BD + import Database.SQLite.Simple ( FromRow, ToRow ) import Dhall ( FromDhall(..), ToDhall(..), auto, input ) @@ -54,19 +60,23 @@ import Relude hiding ( show ) -import Text.Printf ( printf ) +import Utils ( hash ) {-# SPECIALISE hentaiHeader :: [ ( HeaderName, Text ) ] #-} {-# SPECIALISE hentaiHeader :: [ ( HeaderName, ByteString ) ] #-} hentaiHeader :: IsString a => [ ( HeaderName, a ) ] hentaiHeader = [ ( "Connection", "close" ) - , ( "User-Agent", "Hentai@Home 176" ) + , ( "User-Agent", "Hentai@Home 1.6.6" ) , ( "Cache-Control", "public, max-age=31536000" ) - , ( "Server", "Genetic Lifeform and Distributed Open Server 1.6.4" ) + , ( "Server", "Genetic Lifeform and Distributed Open Server 1.6.6" ) , ( "X-Content-Type-Options", "nosniff" ) ] +-- | The protocol level this client implements (official 1.6.6), sent as clientbuild. +clientBuild :: Text +clientBuild = "181" + data ClientProxy = ClientProxy { host :: {-# UNPACK #-} !Text , port :: {-# UNPACK #-} !Integer @@ -98,7 +108,7 @@ = ClientConfig { clientId :: {-# UNPACK #-} !Text , key :: {-# UNPACK #-} !Text - , version :: {-# UNPACK #-} !Text + , version :: {-# UNPACK #-} !Text -- ^ unused, kept so existing configs load; see 'clientBuild' , proxy :: {-# UNPACK #-} !(Maybe ClientProxy) , downloadDir :: {-# UNPACK #-} !Text , cachePath :: {-# UNPACK #-} !Text @@ -143,6 +153,8 @@ , checkIPOrigin :: {-# UNPACK #-} !Bool , floodControl :: {-# UNPACK #-} !Bool , staticRanges :: {-# UNPACK #-} !(HashSet ShortByteString) + , rpcServerIPs :: ![ ByteString ] + -- ^ As sent by the server, e.g. "::ffff:192.0.2.10" } deriving ( Show ) @@ -161,6 +173,7 @@ , checkIPOrigin = True , floodControl = True , staticRanges = mempty + , rpcServerIPs = [] } parseSettings :: [ ByteString ] -> HathSettings @@ -177,6 +190,7 @@ ( "diskremaining_bytes", Just bytes ) -> s { diskRemainingBytes = bytes } ( "static_ranges", _ ) -> s { staticRanges = HashSet.fromList (SBS.toShort <$> BS.split ';' v) } + ( "rpc_server_ip", _ ) -> s { rpcServerIPs = BS.split ';' v } _ -> s) defaultHathSettings data GalleryFile @@ -184,7 +198,7 @@ , galleryFileIndex :: {-# UNPACK #-} !Int , galleryFileName :: {-# UNPACK #-} !ByteString , galleryFileXRes :: {-# UNPACK #-} !ByteString - , galleryFileHash :: {-# UNPACK #-} !ByteString + , galleryFileHash :: !(Maybe ByteString) -- ^ Nothing: generated on demand, unverifiable , galleryFileExt :: {-# UNPACK #-} !ByteString } deriving ( Eq ) @@ -198,48 +212,55 @@ } deriving ( Eq ) -emptyMetadata :: GalleryMetadata -emptyMetadata - = GalleryMetadata - { galleryID = 0 - , galleryFileCount = 0 - , galleryMinXRes = "" - , galleryTitle = "" - , galleryFileList = [] - } +-- | Parse a /16/dl fetchqueue reply (hathdl format). Left carries the reason it was +-- refused: NO_PENDING_DOWNLOADS and other non-lists, or a list failing the 1.6.6 checks. +-- The file list must carry DLHASH = sha1(client key ++ "\n" ++ line, per file line), +-- and names/types must be ones the gallery system can produce, which keeps a forged +-- list from writing outside the download directory. +parseMetadata :: ByteString -> ByteString -> Either Text GalleryMetadata +parseMetadata clientKey bytes = case break (== "FILELIST") (BS.split '\n' bytes) of + ( _, [] ) -> Left $ decodeUtf8 $ BS.strip $ BS.take 64 bytes + ( header, _ : rest ) -> do + let fileLines = filter (not . BS.null) (takeWhile (/= "INFORMATION") rest) + field k = listToMaybe [ BS.drop 1 v | l <- header, let ( f, v ) = BS.break (== ' ') l, f == k ] + gid <- maybeToRight "invalid GID" $ mfilter (> 0) $ readIntExact =<< field "GID" + minXRes <- maybeToRight "invalid MINXRES" $ mfilter validXRes $ field "MINXRES" + when (field "DLHASH" /= Just (hash (clientKey <> foldMap ("\n" <>) fileLines))) + $ Left "DLHASH does not match the file list" + files <- traverse parseFile fileLines + pure + GalleryMetadata + { galleryID = gid + , galleryFileCount = length files + , galleryMinXRes = minXRes + , galleryTitle = fromMaybe "" (field "TITLE") + , galleryFileList = files + } + where + parseFile line = case BS.split ' ' line of + [ page, index, xres, sha1, ext, name ] + | Just p <- readIntExact page + , Just ix <- readIntExact index + , validName name + , ext `elem` [ "jpg", "gif", "png", "webp" ] -> Right + GalleryFile + { galleryFilePage = p + , galleryFileIndex = ix + , galleryFileName = name + , galleryFileXRes = xres + , galleryFileHash = if sha1 == "unknown" then Nothing else Just sha1 + , galleryFileExt = ext + } + _ -> Left $ "invalid file line: " <> decodeUtf8 line -{-# NOINLINE emptyMetadata #-} + validName n = not (BS.null n) && BS.length n <= 250 && not (".." `BS.isInfixOf` n) + && BS.all (\c -> isAsciiUpper c || isAsciiLower c || isDigit c || c `elem` ("._-" :: String)) n -parseMetadata :: ByteString -> GalleryMetadata -parseMetadata bytes = foldl' go emptyMetadata (BS.lines bytes) - where - go metadata line = case BS.words line of - [] -> metadata - [ "GID", maybeGid ] - -> maybe metadata (\( gid, _ ) -> metadata { galleryID = gid }) (BS.readInt maybeGid) - [ "FILECOUNT", maybeCount ] -> maybe - metadata - (\( count, _ ) -> metadata { galleryFileCount = count }) - (BS.readInt maybeCount) - [ "MINXRES", xres ] -> metadata { galleryMinXRes = xres } - ("TITLE" : rest) -> metadata { galleryTitle = BS.unwords rest } - [ maybePage, maybeFid, mxres, hash, ext, basename ] - -> case ( BS.readInt maybePage, BS.readInt maybeFid ) of - ( Just ( page, _ ), Just ( fid, _ ) ) -> metadata - { galleryFileList = GalleryFile - { galleryFilePage = page - , galleryFileIndex = fid - , galleryFileName = basename - , galleryFileXRes = mxres - , galleryFileHash = hash - , galleryFileExt = ext - } - : galleryFileList metadata - } - _ -> metadata - _ -> metadata + validXRes x = x == "org" || (not (BS.null x) && BS.all isDigit x) - {-# INLINE go #-} + readIntExact s = case BS.readInt s of + Just ( n, "" ) -> Just n + _ -> Nothing data FileURI = FileURI { fileHash :: {-# UNPACK #-} !ByteString @@ -250,9 +271,24 @@ } deriving ( Ord, Eq ) +renderFileURIBytes :: FileURI -> ByteString +renderFileURIBytes (FileURI { fileHash, fileSize, fileXRes, fileYRes, fileExt }) + = BD.toStrictByteString + (BD.byteString fileHash + <> "-" + <> BD.intDec fileSize + <> "-" + <> BD.intDec fileXRes + <> "-" + <> BD.intDec fileYRes + <> "-" + <> BD.byteString fileExt) + +renderFileURIText :: FileURI -> Text +renderFileURIText = decodeUtf8 . renderFileURIBytes + instance Show FileURI where - show (FileURI { fileHash, fileSize, fileXRes, fileYRes, fileExt }) - = printf "%s-%d-%d-%d-%s" (BS.unpack fileHash) fileSize fileXRes fileYRes (BS.unpack fileExt) + show = toString . renderFileURIText emptyFileURI :: FileURI emptyFileURI = FileURI { fileHash = "", fileSize = 0, fileXRes = 0, fileYRes = 0, fileExt = "" } @@ -320,6 +356,22 @@ | statusCode response == "OK" = Right $ payload response | otherwise = Left $ RequestFailure $ decodeUtf8 $ statusCode response +-- | Check the 1.6.6 RPC signature. A signed reply starts with OK-