diff --git a/hadoop-rpc.cabal b/hadoop-rpc.cabal --- a/hadoop-rpc.cabal +++ b/hadoop-rpc.cabal @@ -1,5 +1,5 @@ name: hadoop-rpc -version: 1.0.0.1 +version: 1.1.0.0 synopsis: Use the Hadoop RPC interface from Haskell. @@ -16,7 +16,7 @@ homepage: http://github.com/jystic/hadoop-rpc license: Apache-2.0 license-file: LICENSE -author: Jacob Stanley, Conrad Parker +author: Jacob Stanley, Conrad Parker, Luke Clifton maintainer: Jacob Stanley category: Data build-type: Simple @@ -41,8 +41,10 @@ Network.Hadoop.Hdfs Network.Hadoop.Read Network.Hadoop.Rpc + Network.Hadoop.Sasl Network.Hadoop.Socket Network.Hadoop.Stream + Network.Hadoop.Types build-depends: base >= 4.7 && < 5 @@ -50,6 +52,7 @@ , bytestring >= 0.10 , cereal >= 0.4 , exceptions >= 0.6 + , gsasl >= 0.3.6 , hashable >= 1.2.1 , monad-loops >= 0.4 , network >= 2.5 @@ -58,7 +61,7 @@ , socks >= 0.5 , stm >= 2.4 , text >= 1.1 - , transformers >= 0.4 + , transformers >= 0.3 , unix >= 2.7 , unordered-containers >= 0.2 , uuid >= 1.3.4 @@ -80,7 +83,7 @@ buildable: False build-depends: - base >= 4.7 && < 5 + base >= 4.7 && < 5 , hadoop-rpc , protobuf , tasty diff --git a/src/Data/Hadoop/Configuration.hs b/src/Data/Hadoop/Configuration.hs --- a/src/Data/Hadoop/Configuration.hs +++ b/src/Data/Hadoop/Configuration.hs @@ -2,13 +2,17 @@ {-# LANGUAGE RecordWildCards #-} module Data.Hadoop.Configuration - ( getHadoopConfig + ( authUser + , getHadoopConfig , getHadoopUser , getNameNodes + , readPrincipal + , writePrincipal ) where -import Control.Applicative ((<$>), (<*>)) +import Control.Applicative import Control.Exception (IOException, handle) + import qualified Data.ByteString.Char8 as B import qualified Data.HashMap.Lazy as H import Data.Maybe (fromMaybe, mapMaybe) @@ -19,17 +23,23 @@ import System.Environment (lookupEnv) import System.Posix.User (getEffectiveUserName) -import Text.XmlHtml +import Text.XmlHtml (Document, parseXML, docContent) +import Text.XmlHtml (nodeText, childElementTag, descendantElementsTag) + +import Prelude + import Data.Hadoop.Types ------------------------------------------------------------------------ getHadoopConfig :: IO HadoopConfig getHadoopConfig = do - hcUser <- getHadoopUser + udUser <- getHadoopUser + let udAuthUser = Nothing hcNameNodes <- getNameNodes let hcProxy = Nothing + let hcUser = UserDetails{..} return HadoopConfig{..} ------------------------------------------------------------------------ @@ -45,6 +55,28 @@ ------------------------------------------------------------------------ +-- Extract the name to be used for authentication +authUser :: UserDetails -> User +authUser UserDetails{..} = maybe udUser id udAuthUser + +------------------------------------------------------------------------ + +readPrincipal :: Text -> HostName -> Maybe Principal +readPrincipal p host = + case T.split (`elem` ['/', '@']) p of + [pService,"_HOST",pRealm] -> let pHost = host in Just Principal{..} + [pService,pHost,pRealm] -> Just Principal{..} + _ -> + case T.split (`elem` ['@']) p of + [pService,pRealm] -> let pHost = "" in Just Principal{..} + _ -> Nothing + +writePrincipal :: Principal -> Text +writePrincipal Principal{..} = case pHost of + "" -> pService <> "@" <> pRealm + _ -> pService <> "/" <> pHost <> "@" <> pRealm + +------------------------------------------------------------------------ type HadoopXml = H.HashMap Text Text getNameNodes :: IO [NameNode] @@ -54,10 +86,11 @@ return $ fromMaybe [] $ resolveNameNode cfg <$> (stripProto =<< H.lookup fsDefaultNameKey cfg) where - proto = "hdfs://" - fsDefaultNameKey = "fs.defaultFS" - nameNodesPrefix = "dfs.ha.namenodes." - rpcAddressPrefix = "dfs.namenode.rpc-address." + proto = "hdfs://" + fsDefaultNameKey = "fs.defaultFS" + nameNodesPrefix = "dfs.ha.namenodes." + rpcAddressPrefix = "dfs.namenode.rpc-address." + namenodePrincipal = "dfs.namenode.kerberos.principal" stripProto :: Text -> Maybe Text stripProto uri | proto `T.isPrefixOf` uri = Just (T.drop (T.length proto) uri) @@ -65,9 +98,23 @@ resolveNameNode :: HadoopXml -> Text -> [NameNode] resolveNameNode cfg name = case parseEndpoint name of - Just ep -> [ep] -- contains "host:port" directly - Nothing -> mapMaybe (\nn -> lookupAddress cfg $ name <> "." <> nn) - (lookupNameNodes cfg name) + -- contains "host:port" directly + Just ep@Endpoint{..} -> + [ NameNode + { nnEndPoint = ep + , nnPrincipal = lookupPrincipal cfg epHost + } + ] + Nothing -> mapMaybe (\nn -> do + ep <- lookupAddress cfg $ name <> "." <> nn + let pr = lookupPrincipal cfg (epHost ep) + return $ NameNode ep pr + ) (lookupNameNodes cfg name) + + lookupPrincipal :: HadoopXml -> HostName -> Maybe Principal + lookupPrincipal cfg host = do + p <- H.lookup namenodePrincipal cfg + readPrincipal p host lookupNameNodes :: HadoopXml -> Text -> [Text] lookupNameNodes cfg name = fromMaybe [] diff --git a/src/Data/Hadoop/Types.hs b/src/Data/Hadoop/Types.hs --- a/src/Data/Hadoop/Types.hs +++ b/src/Data/Hadoop/Types.hs @@ -12,9 +12,18 @@ ------------------------------------------------------------------------ -type NameNode = Endpoint +data NameNode = NameNode + { nnEndPoint :: !Endpoint + , nnPrincipal :: !(Maybe Principal) + } deriving (Eq, Ord, Show) type SocksProxy = Endpoint +data Principal = Principal + { pService :: !Text + , pHost :: !Text + , pRealm :: !Text + } deriving (Eq, Ord, Show) + data Endpoint = Endpoint { epHost :: !HostName , epPort :: !Port @@ -26,9 +35,14 @@ ------------------------------------------------------------------------ data HadoopConfig = HadoopConfig - { hcUser :: !User + { hcUser :: !UserDetails , hcNameNodes :: ![NameNode] , hcProxy :: !(Maybe SocksProxy) + } deriving (Eq, Ord, Show) + +data UserDetails = UserDetails + { udUser :: !User + , udAuthUser :: !(Maybe User) } deriving (Eq, Ord, Show) ------------------------------------------------------------------------ diff --git a/src/Network/Hadoop/Hdfs.hs b/src/Network/Hadoop/Hdfs.hs --- a/src/Network/Hadoop/Hdfs.hs +++ b/src/Network/Hadoop/Hdfs.hs @@ -25,27 +25,29 @@ , setPermissions ) where -import Control.Applicative (Applicative(..), (<$>)) +import Control.Applicative import Control.Concurrent.STM import Control.Exception (SomeException(..), throw) import Control.Monad (ap, when) import Control.Monad.Catch (MonadMask(..), MonadThrow(..), MonadCatch(..)) import Control.Monad.IO.Class (MonadIO(..)) + import Data.ByteString (ByteString) import qualified Data.ByteString.Char8 as B import Data.Maybe (fromMaybe) import Data.Monoid ((<>)) +import Data.ProtocolBuffers +import Data.ProtocolBuffers.Orphans () import Data.Text (Text) import qualified Data.Text as T import qualified Data.Text.Encoding as T import qualified Data.Vector as V import Data.Word (Word32) +import Prelude + import qualified Data.Hadoop.Protobuf.ClientNameNode as P import qualified Data.Hadoop.Protobuf.Hdfs as P -import Data.ProtocolBuffers -import Data.ProtocolBuffers.Orphans () - import Data.Hadoop.Configuration import Data.Hadoop.HdfsPath import Data.Hadoop.Types @@ -103,10 +105,10 @@ runHdfs' config hdfs runHdfs' :: HadoopConfig -> Hdfs a -> IO a -runHdfs' config@HadoopConfig{..} hdfs = S.bracketSocket hcProxy nameNode session +runHdfs' config@HadoopConfig{..} hdfs = S.bracketSocket hcProxy (nnEndPoint nameNode) session where session socket = do - conn <- initConnectionV9 config hdfsProtocol socket + conn <- initConnectionV9 config nameNode hdfsProtocol socket unHdfs hdfs conn nameNode = case hcNameNodes of diff --git a/src/Network/Hadoop/Read.hs b/src/Network/Hadoop/Read.hs --- a/src/Network/Hadoop/Read.hs +++ b/src/Network/Hadoop/Read.hs @@ -3,6 +3,29 @@ {-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE TupleSections #-} +{-| + + Example for accessing HDFS files: + +>import qualified Data.ByteString.Char8 as BC +>import Network.Hadoop.Hdfs +>import Network.Hadoop.Read +>import System.IO (IOMode(..), withBinaryFile) +> +> main :: IO () +> main = do +> (Just readHandle) <- runHdfs $ openRead $ BC.pack +> withBinaryFile "local-file" WriteMode $ \handle -> +> withHdfsReader (BC.hPut handle) "/absolute/path/to/file" +> +> withHdfsReader :: (BC.ByteString -> IO ()) -> String -> IO () +> withHdfsReader action path = do +> readHandle_ <- runHdfs $ openRead $ BC.pack path +> case readHandle_ of +> (Just readHandle) -> hdfsMapM_ action readHandle +> Nothing -> error "no read handle + +-} module Network.Hadoop.Read ( HdfsReadHandle , openRead @@ -13,26 +36,30 @@ , hdfsCat ) where -import Control.Applicative ((<$>), (<$)) +import Control.Applicative import Control.Exception (IOException, throwIO) import Control.Monad (foldM) import Control.Monad.Catch (MonadMask, catch) import Control.Monad.IO.Class (MonadIO(..)) import Control.Monad.Loops (iterateUntilM) -import Control.Monad.Trans.State import Control.Monad.Trans.Class (lift) +import Control.Monad.Trans.State (StateT(..), execStateT, get, put) + import Data.ByteString (ByteString) import qualified Data.ByteString.Char8 as B import Data.Maybe (fromMaybe) import Data.ProtocolBuffers import Data.ProtocolBuffers.Orphans () +import qualified Data.Serialize.Get as Get +import qualified Data.Serialize.Put as Put import qualified Data.Text as T import qualified Data.Text.Encoding as T import Data.Word (Word64) -import qualified Data.Serialize.Get as Get -import qualified Data.Serialize.Put as Put +import Text.Printf (printf) +import Prelude hiding (rem) + import Data.Hadoop.Protobuf.ClientNameNode import Data.Hadoop.Protobuf.DataTransfer import Data.Hadoop.Protobuf.Hdfs @@ -41,9 +68,6 @@ import Network.Hadoop.Rpc import qualified Network.Hadoop.Socket as S import qualified Network.Hadoop.Stream as Stream -import Text.Printf (printf) - -import Prelude hiding (rem) ------------------------------------------------------------------------ diff --git a/src/Network/Hadoop/Rpc.hs b/src/Network/Hadoop/Rpc.hs --- a/src/Network/Hadoop/Rpc.hs +++ b/src/Network/Hadoop/Rpc.hs @@ -14,50 +14,41 @@ , invoke ) where -import Control.Applicative ((<$>)) +import Control.Applicative import Control.Concurrent (ThreadId, forkIO, newEmptyMVar, putMVar, takeMVar) import Control.Concurrent.STM import Control.Exception (SomeException(..), throwIO, handle) import Control.Monad (forever, when) +import Control.Monad.IO.Class (liftIO) + import Data.ByteString (ByteString) import qualified Data.ByteString.Lazy as L import qualified Data.HashMap.Strict as H import Data.Hashable (Hashable) import Data.Maybe (fromMaybe, isNothing) -import Data.Monoid ((<>)) -import Data.Monoid (mempty) -import Data.Text (Text) -import qualified Data.Text as T -import qualified Data.UUID as UUID -import System.Random (randomIO) - +import Data.Monoid import Data.ProtocolBuffers import Data.ProtocolBuffers.Orphans () import Data.Serialize.Get import Data.Serialize.Put +import Data.Text (Text) +import qualified Data.Text as T +import qualified Data.Text.Encoding as T +import qualified Data.UUID as UUID -import qualified Data.Hadoop.Protobuf.Headers as P -import Data.Hadoop.Types -import qualified Network.Hadoop.Stream as S +import qualified Network.Protocol.SASL.GNU as GS import Network.Socket (Socket) ------------------------------------------------------------------------- - -data Connection = Connection - { cnVersion :: !Int - , cnConfig :: !HadoopConfig - , cnProtocol :: !Protocol - , invokeRaw :: !(Method -> RawRequest -> (RawResponse -> IO ()) -> IO ()) - } +import System.Random (randomIO) -data Protocol = Protocol - { prName :: !Text - , prVersion :: !Int - } deriving (Eq, Ord, Show) +import Prelude -type Method = Text -type RawRequest = ByteString -type RawResponse = Either SomeException ByteString +import Data.Hadoop.Configuration +import qualified Data.Hadoop.Protobuf.Headers as P +import Data.Hadoop.Types +import Network.Hadoop.Sasl +import qualified Network.Hadoop.Stream as S +import Network.Hadoop.Types type CallId = Int @@ -77,8 +68,8 @@ , csFatalError :: !(TVar (Maybe SomeException)) } -initConnectionV9 :: HadoopConfig -> Protocol -> Socket -> IO Connection -initConnectionV9 config@HadoopConfig{..} protocol sock = do +initConnectionV9 :: HadoopConfig -> NameNode -> Protocol -> Socket -> IO Connection +initConnectionV9 config@HadoopConfig{..} nameNode protocol sock = do csStream <- S.mkSocketStream sock csClientId <- mkClientId @@ -86,7 +77,14 @@ putByteString "hrpc" putWord8 9 -- version putWord8 0 -- rpc service class (0 = default/protobuf, 1 = built-in, 2 = writable, 3 = protobuf - putWord8 0 -- auth protocol (0 = none, -33/0xDF = sasl) + -- auth protocol (0 = none, -33/0xDF = sasl) + putWord8 $ maybe 0 (const 0xDF) (nnPrincipal nameNode) + + case nnPrincipal nameNode of + Nothing -> return () + Just principal -> saslAuth principal csStream + + S.runPut csStream $ putMessage $ delimitedBytesL (rpcRequestHeaderProto csClientId (-3)) <> delimitedBytesL (contextProto protocol hcUser) @@ -102,6 +100,95 @@ return (Connection 7 config protocol (enqueue cs)) where + invokeSasl :: (Encode a, Decode b) => S.Stream -> a -> IO b + invokeSasl stream msgr = do + let message = delimitedBytesL (rpcRequestHeaderProto (ClientId "") (-33)) + <> delimitedBytesL msgr + + S.runPut stream (putMessage message) + mget <- S.maybeGet stream $ do + n <- fromIntegral <$> getWord32be + -- TODO Would be nice if we didn't have to isolate here + -- TODO and could stream instead. We could stream if we + -- TODO were able to read the varint length prefix + -- TODO ourselves and keep track of how many bytes were + -- TODO remaining instead of calling `getRemaining`. + isolate n $ do + hdr <- decodeLengthPrefixedMessage + msg <- case getField (P.rspStatus hdr) of + P.Success -> Right <$> getRemaining + _ -> return . Left . SomeException $ rspError hdr + return (hdr, msg) + + case mget of + Nothing -> throwIO ConnectionClosed + Just (_, msg) -> + case msg of + Left ex -> throwIO ex + Right m -> case fromDelimitedBytes m of + Left ex -> throwIO ex + Right x -> return x + + saslAuth :: Principal -> S.Stream -> IO () + saslAuth Principal{..} c = do + negResp <- liftIO $ invokeSasl c RpcSaslProto + { spState = putField Negotiate + , spVersion = mempty + , spToken = mempty + , spAuths = mempty + } + case getField $ spState negResp of + Negotiate -> return () + _ -> throwIO $ SomeException FailedNegotiation + + let auths = getField $ spAuths negResp + authsMap = zip (map (getField . saMechanism) auths) auths + + res <- GS.runSASL $ do + -- Select a common mechanism + mSelectedMech <- GS.clientSuggestMechanism $ map (GS.Mechanism . fst) authsMap + GS.Mechanism selectedMech <- liftIO $ + maybe (throwIO $ SomeException NoSharedMechanism) return mSelectedMech + -- Safe because `selectedMech` was selected from one of the keys in `authsMap` + let Just selectedAuth = lookup selectedMech authsMap + GS.runClient (GS.Mechanism selectedMech) $ do + GS.setProperty GS.PropertyService $ T.encodeUtf8 pService + GS.setProperty GS.PropertyHostname $ T.encodeUtf8 pHost + GS.setProperty GS.PropertyAuthID $ T.encodeUtf8 (authUser hcUser) + (token,_) <- GS.step "" + r <- liftIO $ invokeSasl c RpcSaslProto + { spState = putField Initiate + , spVersion = mempty + , spToken = putField $ Just token + , spAuths = putField $ [selectedAuth] + } + saslAuthLoop c r + + case res of + Left err -> throwIO $ SomeException $ SaslException err + Right _ -> return () + + saslAuthLoop :: S.Stream -> RpcSaslProto -> GS.Session () + saslAuthLoop c r = do + case getField $ spState r of + Challenge -> do + let mToken = getField $ spToken r + case mToken of + Nothing -> liftIO $ throwIO $ SomeException NoToken + Just token -> do + (resToken, _) <- GS.step token + resp <- liftIO $ invokeSasl c RpcSaslProto + { spToken = putField $ Just resToken + , spState = putField $ Response + , spVersion = mempty + , spAuths = mempty + } + saslAuthLoop c resp + Success -> return () + _ -> liftIO $ throwIO $ SomeException UnexptededState + + + enqueue :: ConnectionState -> Method -> RawRequest @@ -188,11 +275,11 @@ ------------------------------------------------------------------------ -contextProto :: Protocol -> User -> P.IpcConnectionContext +contextProto :: Protocol -> UserDetails -> P.IpcConnectionContext contextProto protocol user = P.IpcConnectionContext { P.ctxProtocol = putField (Just (prName protocol)) , P.ctxUserInfo = putField (Just P.UserInformation - { P.effectiveUser = putField (Just user) + { P.effectiveUser = putField (Just (authUser user)) , P.realUser = mempty }) } diff --git a/src/Network/Hadoop/Sasl.hs b/src/Network/Hadoop/Sasl.hs new file mode 100644 --- /dev/null +++ b/src/Network/Hadoop/Sasl.hs @@ -0,0 +1,67 @@ +{-# LANGUAGE DataKinds #-} +{-# LANGUAGE DeriveDataTypeable #-} +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE RecordWildCards #-} + +module Network.Hadoop.Sasl where + +import Network.Protocol.SASL.GNU +import Data.ProtocolBuffers + +import Control.Exception (Exception) + +import GHC.Generics +import Data.ByteString +import Data.Text +import Data.Typeable +import Data.Word + +------------------------------------------------------------------------ + +newtype SaslException = SaslException Error + deriving (Show, Typeable) + +instance Exception SaslException + +data SaslNegotiationError + = FailedNegotiation + | NoSharedMechanism + | NoToken + | UnexptededState + deriving (Show, Typeable) + +instance Exception SaslNegotiationError + +rpcSaslProto :: Text +rpcSaslProto = "RpcSaslProto" + +data RpcSaslProto = RpcSaslProto + { spVersion :: Optional 1 (Value Word32) + , spState :: Required 2 (Enumeration SaslState) + , spToken :: Optional 3 (Value ByteString) + , spAuths :: Repeated 4 (Message SaslAuth) + } deriving (Eq, Ord, Show, Generic) + +data SaslAuth = SaslAuth + { saMethod :: Required 1 (Value ByteString) + , saMechanism :: Required 2 (Value ByteString) + , saProtocol :: Optional 3 (Value ByteString) + , saServerId :: Optional 4 (Value ByteString) + , saChallenge :: Optional 5 (Value ByteString) + } deriving (Eq, Ord, Show, Generic) + +data SaslState + = Success + | Negotiate + | Initiate + | Challenge + | Response + | Wrap + deriving (Eq, Ord, Show, Enum, Bounded) + +instance Encode SaslAuth +instance Decode SaslAuth +instance Encode RpcSaslProto +instance Decode RpcSaslProto + diff --git a/src/Network/Hadoop/Stream.hs b/src/Network/Hadoop/Stream.hs --- a/src/Network/Hadoop/Stream.hs +++ b/src/Network/Hadoop/Stream.hs @@ -17,19 +17,23 @@ , close ) where -import Control.Applicative ((<$>)) +import Control.Applicative import qualified Control.Concurrent.Chan as Chan import Control.Exception (throwIO) import Control.Monad (forM_) + import qualified Data.Attoparsec.ByteString as Atto import qualified Data.ByteString as B import qualified Data.ByteString.Lazy as L import Data.IORef (IORef, newIORef, readIORef, writeIORef) import qualified Data.Serialize.Get as Get import qualified Data.Serialize.Put as Put + import qualified Network.Socket as S import qualified Network.Socket.ByteString as B (recv) import qualified Network.Socket.ByteString.Lazy as L (sendAll) + +import Prelude import Data.Hadoop.Types diff --git a/src/Network/Hadoop/Types.hs b/src/Network/Hadoop/Types.hs new file mode 100644 --- /dev/null +++ b/src/Network/Hadoop/Types.hs @@ -0,0 +1,25 @@ +module Network.Hadoop.Types where + +import Data.Text +import Data.ByteString +import Control.Exception +import Data.Hadoop.Types + +------------------------------------------------------------------------ + +data Connection = Connection + { cnVersion :: !Int + , cnConfig :: !HadoopConfig + , cnProtocol :: !Protocol + , invokeRaw :: !(Method -> RawRequest -> (RawResponse -> IO ()) -> IO ()) + } + +data Protocol = Protocol + { prName :: !Text + , prVersion :: !Int + } deriving (Eq, Ord, Show) + +type Method = Text +type RawRequest = ByteString +type RawResponse = Either SomeException ByteString + diff --git a/test/Main.hs b/test/Main.hs --- a/test/Main.hs +++ b/test/Main.hs @@ -21,7 +21,7 @@ config :: HadoopConfig config = HadoopConfig { - hcUser = "hdfs" - , hcNameNodes = [(Endpoint "127.0.0.1" 8020)] + hcUser = UserDetails "hdfs" Nothing + , hcNameNodes = [NameNode (Endpoint "127.0.0.1" 8020) Nothing] , hcProxy = Just (Endpoint "127.0.0.1" 2080) }