diff --git a/Database/Cassandra/CQL.hs b/Database/Cassandra/CQL.hs --- a/Database/Cassandra/CQL.hs +++ b/Database/Cassandra/CQL.hs @@ -3,7 +3,7 @@ BangPatterns, OverlappingInstances, DataKinds, GADTs, KindSignatures #-} -- | Haskell client for Cassandra's CQL protocol -- --- For examples, take a look at the /tests/ directory in the source archive. +-- For examples, take a look at the /tests/ directory in the source archive. -- -- Here's the correspondence between CQL and Haskell types: -- @@ -115,6 +115,8 @@ CassandraException(..), CassandraCommsError(..), TransportDirection(..), + -- * Auth + Authentication (..), -- * Queries Query, Style(..), @@ -212,7 +214,8 @@ -- | A handle for the state of the connection pool. data Pool = Pool { piKeyspace :: Keyspace, - piSessions :: TVar (Seq Session) + piSessions :: TVar (Seq Session), + piAuth :: Maybe Authentication } class MonadCatchIO m => MonadCassandra m where @@ -234,13 +237,14 @@ getCassandraPool = lift getCassandraPool -- | Construct a pool of Cassandra connections. -newPool :: [Server] -> Keyspace -> IO Pool -newPool svrs ks = do +newPool :: [Server] -> Keyspace -> Maybe Authentication -> IO Pool +newPool svrs ks auth = do let sessions = map (\svr -> Session svr Nothing) svrs sess <- atomically $ newTVar (Seq.fromList sessions) return $ Pool { piKeyspace = ks, - piSessions = sess + piSessions = sess, + piAuth = auth } takeSession :: Pool -> IO Session @@ -256,8 +260,8 @@ putSession :: Pool -> Session -> IO () putSession pool ses = atomically $ modifyTVar (piSessions pool) (|> ses) -connectIfNeeded :: Pool -> Session -> IO Session -connectIfNeeded pool session = +connectIfNeeded :: Pool -> Maybe Authentication -> Session -> IO Session +connectIfNeeded pool auth session = if isJust (sesActive session) then return session else do @@ -282,7 +286,7 @@ actSocket = socket, actQueryCache = M.empty } - active' <- execStateT (introduce pool) active + active' <- execStateT (introduce pool auth) active return $ session { sesActive = Just active' } Nothing -> return session @@ -328,18 +332,18 @@ get = do w <- getWord8 case w of - 0x00 -> return $ ERROR - 0x01 -> return $ STARTUP - 0x02 -> return $ READY - 0x03 -> return $ AUTHENTICATE - 0x04 -> return $ CREDENTIALS - 0x05 -> return $ OPTIONS - 0x06 -> return $ SUPPORTED - 0x07 -> return $ QUERY - 0x08 -> return $ RESULT - 0x09 -> return $ PREPARE - 0x0a -> return $ EXECUTE - 0x0b -> return $ REGISTER + 0x00 -> return $ ERROR + 0x01 -> return $ STARTUP + 0x02 -> return $ READY + 0x03 -> return $ AUTHENTICATE + 0x04 -> return $ CREDENTIALS + 0x05 -> return $ OPTIONS + 0x06 -> return $ SUPPORTED + 0x07 -> return $ QUERY + 0x08 -> return $ RESULT + 0x09 -> return $ PREPARE + 0x0a -> return $ EXECUTE + 0x0b -> return $ REGISTER 0x0c -> return $ EVENT _ -> fail $ "unknown opcode 0x"++showHex w "" @@ -511,6 +515,7 @@ -- bug. data CassandraCommsError = AuthenticationException Text | LocalProtocolError Text Text + | MissingAuthenticationError Text Text | ValueMarshallingException TransportDirection Text Text | CassandraIOException IOException | ShortRead @@ -528,7 +533,7 @@ parseError = do code <- getWord32be case code of - 0x0000 -> ServerError <$> getElt <*> pure qt + 0x0000 -> ServerError <$> getElt <*> pure qt 0x000A -> ProtocolError <$> getElt <*> pure qt 0x0100 -> BadCredentials <$> getElt <*> pure qt 0x1000 -> UnavailableException <$> getElt <*> getElt @@ -553,16 +558,38 @@ 0x2500 -> Unprepared <$> getElt <*> getElt <*> pure qt _ -> fail $ "unknown error code 0x"++showHex code "" -introduce :: Pool -> StateT ActiveSession IO () -introduce pool = do + +type User = T.Text +type Password = T.Text +data Authentication = PasswordAuthenticator User Password +type Credentials = [(Text, Text)] + +authCredentials :: Authentication -> Credentials +authCredentials (PasswordAuthenticator user password) = [("username", user), ("password", password)] + +authenticate :: Authentication -> StateT ActiveSession IO () +authenticate auth = do + let qt = "" + sendFrame $ Frame [] 0 CREDENTIALS $ encodeElt $ authCredentials auth + fr2 <- recvFrame qt + case frOpcode fr2 of + READY -> return () + ERROR -> throwError qt (frBody fr2) + op -> throw $ LocalProtocolError ("introduce: unexpected opcode " `T.append` T.pack (show op)) qt + +introduce :: Pool -> Maybe Authentication -> StateT ActiveSession IO () +introduce pool auth = do let qt = "" sendFrame $ Frame [] 0 STARTUP $ encodeElt $ ([("CQL_VERSION", "3.0.0")] :: [(Text, Text)]) fr <- recvFrame qt case frOpcode fr of - AUTHENTICATE -> throw $ AuthenticationException "authentication not implemented yet" - READY -> return () - ERROR -> throwError qt (frBody fr) - op -> throw $ LocalProtocolError ("introduce: unexpected opcode " `T.append` T.pack (show op)) qt + AUTHENTICATE -> maybe + (throw $ MissingAuthenticationError "introduce: server expects auth but none provided" "") + authenticate auth + READY -> return () + ERROR -> throwError qt (frBody fr) + op -> throw $ LocalProtocolError ("introduce: unexpected opcode " `T.append` T.pack (show op)) qt + let Keyspace ksName = piKeyspace pool let q = query $ "USE " `T.append` ksName :: Query Rows () () res <- executeInternal q () ONE @@ -573,8 +600,9 @@ withSession :: MonadCassandra m => (Pool -> StateT ActiveSession IO a) -> m a withSession code = do pool <- getCassandraPool + let auth = piAuth pool mA <- liftIO $ do - session <- connectIfNeeded pool =<< takeSession pool + session <- connectIfNeeded pool auth =<< takeSession pool case sesActive session of Just active -> do (a, active') <- runStateT (code pool) active @@ -595,7 +623,7 @@ return Nothing case mA of Just a -> return a - Nothing -> withSession code -- Try again until we succeed + Nothing -> withSession code -- Try again until we succeed -- | The name of a Cassandra keyspace. See the Cassandra documentation for more -- information. @@ -606,7 +634,7 @@ newtype Table = Table Text deriving (Eq, Ord, Show, IsString, ProtoElt) --- | A fully qualified identification of a table that includes the 'Keyspace'. +-- | A fully qualified identification of a table that includes the 'Keyspace'. data TableSpec = TableSpec Keyspace Table deriving (Eq, Ord, Show) @@ -971,7 +999,7 @@ newtype QueryID = QueryID (Digest SHA1) deriving (Eq, Ord, Show) --- | The first type argument for Query. Tells us what kind of query it is. +-- | The first type argument for Query. Tells us what kind of query it is. data Style = Schema -- ^ A query that modifies the schema, such as DROP TABLE or CREATE TABLE | Write -- ^ A query that writes data, such as an INSERT or UPDATE | Rows -- ^ A query that returns a list of rows, such as SELECT @@ -1105,17 +1133,17 @@ where ba = casObliterate a . encodeCas $ a encodeNested !i (a, _) (ta:_) = Left $ Mismatch i (casType a) ta - encodeNested !i vs [] = Left $ WrongNumber (i + countNested vs) i + encodeNested !i vs [] = Left $ WrongNumber (i + countNested vs) i decodeNested !i ((ta, mba):rem) | ta `equivalent` casType (undefined :: a) = case (decodeCas <$> mba, casType (undefined :: a), decodeNested (i+1) rem) of (Nothing, CMaybe _, Right arem) -> Right (casNothing, arem) (Nothing, _, _) -> Left $ NullValue i ta (Just (Left err), _, _) -> Left $ DecodeFailure i err - (_, _, Left err) -> Left err + (_, _, Left err) -> Left err (Just (Right a), _, Right arem) -> Right (a, arem) decodeNested !i ((ta, _):rem) = Left $ Mismatch i (casType (undefined :: a)) ta decodeNested !i [] = Left $ WrongNumber (i + 1 + countNested (undefined :: rem)) i - countNested _ = let n = 1 + countNested (undefined :: rem) + countNested _ = let n = 1 + countNested (undefined :: rem) in seq n n -- | A type class for a tuple of 'CasType' instances, representing either a list of @@ -1306,10 +1334,10 @@ 0x0007 -> pure EACH_QUORUM _ -> fail $ "unknown consistency value 0x"++showHex w "" --- | A low-level function in case you need some rarely-used capabilities. +-- | A low-level function in case you need some rarely-used capabilities. executeRaw :: (MonadCassandra m, CasValues i) => Query style any_i any_o -> i -> Consistency -> m (Result [Maybe ByteString]) -executeRaw query i cons = withSession $ \_ -> executeInternal query i cons +executeRaw query i cons = withSession (\_ -> executeInternal query i cons) executeInternal :: CasValues values => Query style any_i any_o -> values -> Consistency -> StateT ActiveSession IO (Result [Maybe ByteString]) @@ -1326,7 +1354,7 @@ Nothing -> putWord32be 0xffffffff Just value -> do let enc = encodeCas value - putWord32be (fromIntegral $ B.length enc) + putWord32be (fromIntegral $ B.length enc) putByteString enc putElt cons fr <- recvFrame (queryText query) @@ -1404,5 +1432,4 @@ -- | Execute Cassandra queries. runCas :: Pool -> Cas a -> IO a -runCas pool (Cas code) = runReaderT code pool - +runCas pool (Cas code) = runReaderT code pool diff --git a/cassandra-cql.cabal b/cassandra-cql.cabal --- a/cassandra-cql.cabal +++ b/cassandra-cql.cabal @@ -1,15 +1,16 @@ name: cassandra-cql -version: 0.3.0.1 +version: 0.4.0.1 synopsis: Haskell client for Cassandra's CQL protocol description: Haskell client for Cassandra's CQL protocol . Revision history: 0.3.0.1 Fix socket issue on Mac. + 0.4.0.1 Add PasswordAuthenticator (thanks Curtis Carter) & accept ghc-7.8 license: BSD3 license-file: LICENSE author: Stephen Blackheath maintainer: http://blacksapphire.com/antispam/ -copyright: (c) Stephen Blackheath 2013 +copyright: (c) Stephen Blackheath 2013-2014 category: Database build-type: Simple stability: alpha @@ -32,7 +33,7 @@ library exposed-modules: Database.Cassandra.CQL - build-depends: base >= 4.5.0.0 && < 4.7.0.0, + build-depends: base >= 4.5.0.0 && <= 4.8.0.0, containers >= 0.4.0.0, mtl >= 2.1.0, MonadCatchIO-transformers >= 0.3.0.0, @@ -48,4 +49,3 @@ ghc-options: -Wall -fno-warn-name-shadowing -fno-warn-unused-matches -fno-warn-missing-signatures -fno-warn-orphans -fno-warn-unused-imports -fno-warn-unused-binds - diff --git a/tests/example.hs b/tests/example.hs --- a/tests/example.hs +++ b/tests/example.hs @@ -29,14 +29,17 @@ ignoreDropFailure :: Cas () -> Cas () ignoreDropFailure code = code `catch` \exc -> case exc of ConfigError _ _ -> return () -- Ignore the error if the table doesn't exist + Invalid _ _ -> return () _ -> throw exc main = do + --let auth = Just (PasswordAuthenticator "cassandra" "cassandra") + let auth = Nothing {- Assuming a 'test' keyspace already exists. Here's some CQL to create it: CREATE KEYSPACE test WITH replication = { 'class' : 'SimpleStrategy', 'replication_factor' : '1' }; -} - pool <- newPool [("localhost", "9042")] "test" -- servers, keyspace + pool <- newPool [("localhost", "9042")] "test" auth -- servers, keyspace, maybe auth runCas pool $ do ignoreDropFailure $ liftIO . print =<< executeSchema QUORUM dropSongs () liftIO . print =<< executeSchema QUORUM createSongs () diff --git a/tests/test-decimal.hs b/tests/test-decimal.hs --- a/tests/test-decimal.hs +++ b/tests/test-decimal.hs @@ -26,10 +26,13 @@ ignoreDropFailure :: Cas () -> Cas () ignoreDropFailure code = code `catch` \exc -> case exc of ConfigError _ _ -> return () -- Ignore the error if the table doesn't exist + Invalid _ _ -> return () _ -> throw exc main = do - pool <- newPool [("localhost", "9042")] "test" -- servers, keyspace + --let auth = Just (PasswordAuthenticator "cassandra" "cassandra") + let auth = Nothing + pool <- newPool [("localhost", "9042")] "test" auth -- servers, keyspace, auth runCas pool $ do ignoreDropFailure $ liftIO . print =<< executeSchema QUORUM dropLists () liftIO . print =<< executeSchema QUORUM createLists () @@ -50,8 +53,7 @@ executeWrite QUORUM insert (u6, read "-3.141592654") executeWrite QUORUM insert (u7, read "-0.000000001") executeWrite QUORUM insert (u8, read "118000") - + let us = [u1,u2,u3,u4,u5,u6,u7, u8] forM_ us $ \u -> liftIO . print =<< executeRow QUORUM select u - diff --git a/tests/test-double.hs b/tests/test-double.hs --- a/tests/test-double.hs +++ b/tests/test-double.hs @@ -29,10 +29,13 @@ ignoreDropFailure :: Cas () -> Cas () ignoreDropFailure code = code `catch` \exc -> case exc of ConfigError _ _ -> return () -- Ignore the error if the table doesn't exist + Invalid _ _ -> return () _ -> throw exc main = do - pool <- newPool [("localhost", "9042")] "test" -- servers, keyspace + --let auth = Just (PasswordAuthenticator "cassandra" "cassandra") + let auth = Nothing + pool <- newPool [("localhost", "9042")] "test" auth -- servers, keyspace, auth runCas pool $ do ignoreDropFailure $ liftIO . print =<< executeSchema QUORUM dropLists () liftIO . print =<< executeSchema QUORUM createLists () @@ -45,4 +48,3 @@ executeWrite QUORUM insert (u3, 3.141592654) liftIO . print =<< executeRows QUORUM select () - diff --git a/tests/test-float.hs b/tests/test-float.hs --- a/tests/test-float.hs +++ b/tests/test-float.hs @@ -29,10 +29,13 @@ ignoreDropFailure :: Cas () -> Cas () ignoreDropFailure code = code `catch` \exc -> case exc of ConfigError _ _ -> return () -- Ignore the error if the table doesn't exist + Invalid _ _ -> return () _ -> throw exc main = do - pool <- newPool [("localhost", "9042")] "test" -- servers, keyspace + --let auth = Just (PasswordAuthenticator "cassandra" "cassandra") + let auth = Nothing + pool <- newPool [("localhost", "9042")] "test" auth -- servers, keyspace, auth runCas pool $ do ignoreDropFailure $ liftIO . print =<< executeSchema QUORUM dropLists () liftIO . print =<< executeSchema QUORUM createLists () @@ -45,4 +48,3 @@ executeWrite QUORUM insert (u3, 3.141592654) liftIO . print =<< executeRows QUORUM select () - diff --git a/tests/test-inet.hs b/tests/test-inet.hs --- a/tests/test-inet.hs +++ b/tests/test-inet.hs @@ -28,10 +28,13 @@ ignoreDropFailure :: Cas () -> Cas () ignoreDropFailure code = code `catch` \exc -> case exc of ConfigError _ _ -> return () -- Ignore the error if the table doesn't exist + Invalid _ _ -> return () _ -> throw exc main = do - pool <- newPool [("localhost", "9042")] "test" -- servers, keyspace + --let auth = Just (PasswordAuthenticator "cassandra" "cassandra") + let auth = Nothing + pool <- newPool [("localhost", "9042")] "test" auth -- servers, keyspace, auth runCas pool $ do ignoreDropFailure $ liftIO . print =<< executeSchema QUORUM dropLists () liftIO . print =<< executeSchema QUORUM createLists () @@ -44,4 +47,3 @@ executeWrite QUORUM insert (u2, a2) liftIO . print =<< executeRows QUORUM select () - diff --git a/tests/test-list.hs b/tests/test-list.hs --- a/tests/test-list.hs +++ b/tests/test-list.hs @@ -39,10 +39,13 @@ ignoreDropFailure :: Cas () -> Cas () ignoreDropFailure code = code `catch` \exc -> case exc of ConfigError _ _ -> return () -- Ignore the error if the table doesn't exist + Invalid _ _ -> return () _ -> throw exc main = do - pool <- newPool [("localhost", "9042")] "test" -- servers, keyspace + --let auth = Just (PasswordAuthenticator "cassandra" "cassandra") + let auth = Nothing + pool <- newPool [("localhost", "9042")] "test" auth -- servers, keyspace, auth runCas pool $ do do ignoreDropFailure $ liftIO . print =<< executeSchema QUORUM dropListsI () @@ -54,19 +57,18 @@ executeWrite QUORUM insertI (u1, [10,11,12]) executeWrite QUORUM insertI (u2, [2,4,6,8]) executeWrite QUORUM insertI (u3, [900,1000,1100]) - + liftIO . print =<< executeRows QUORUM selectI () do ignoreDropFailure $ liftIO . print =<< executeSchema QUORUM dropListsT () liftIO . print =<< executeSchema QUORUM createListsT () - + u1 <- liftIO randomIO u2 <- liftIO randomIO u3 <- liftIO randomIO executeWrite QUORUM insertT (u1, ["dog","cat","rabbit"]) executeWrite QUORUM insertT (u2, ["carrot","tomato"]) executeWrite QUORUM insertT (u3, ["a","b","c"]) - - liftIO . print =<< executeRows QUORUM selectT () + liftIO . print =<< executeRows QUORUM selectT () diff --git a/tests/test-map.hs b/tests/test-map.hs --- a/tests/test-map.hs +++ b/tests/test-map.hs @@ -29,10 +29,13 @@ ignoreDropFailure :: Cas () -> Cas () ignoreDropFailure code = code `catch` \exc -> case exc of ConfigError _ _ -> return () -- Ignore the error if the table doesn't exist + Invalid _ _ -> return () _ -> throw exc main = do - pool <- createCassandraPool [("localhost", "9042")] "test" -- servers, keyspace + --let auth = Just (PasswordAuthenticator "cassandra" "cassandra") + let auth = Nothing + pool <- newPool [("localhost", "9042")] "test" auth -- servers, keyspace, auth runCas pool $ do ignoreDropFailure $ liftIO . print =<< executeSchema QUORUM dropLists () liftIO . print =<< executeSchema QUORUM createLists () @@ -45,4 +48,3 @@ executeWrite QUORUM insert (u3, M.fromList [(12, "dozen")]) liftIO . print =<< executeRows QUORUM select () - diff --git a/tests/test-set.hs b/tests/test-set.hs --- a/tests/test-set.hs +++ b/tests/test-set.hs @@ -29,10 +29,13 @@ ignoreDropFailure :: Cas () -> Cas () ignoreDropFailure code = code `catch` \exc -> case exc of ConfigError _ _ -> return () -- Ignore the error if the table doesn't exist + Invalid _ _ -> return () _ -> throw exc main = do - pool <- newPool [("localhost", "9042")] "test" -- servers, keyspace + --let auth = Just (PasswordAuthenticator "cassandra" "cassandra") + let auth = Nothing + pool <- newPool [("localhost", "9042")] "test" auth -- servers, keyspace, auth runCas pool $ do ignoreDropFailure $ liftIO . print =<< executeSchema QUORUM dropLists () liftIO . print =<< executeSchema QUORUM createLists () @@ -45,4 +48,3 @@ executeWrite QUORUM insert (u3, S.fromList ["dozen"]) liftIO . print =<< executeRows QUORUM select () - diff --git a/tests/test-timestamp.hs b/tests/test-timestamp.hs --- a/tests/test-timestamp.hs +++ b/tests/test-timestamp.hs @@ -26,10 +26,13 @@ ignoreDropFailure :: Cas () -> Cas () ignoreDropFailure code = code `catch` \exc -> case exc of ConfigError _ _ -> return () -- Ignore the error if the table doesn't exist + Invalid _ _ -> return () _ -> throw exc main = do - pool <- newPool [("localhost", "9042")] "test" -- servers, keyspace + --let auth = Just (PasswordAuthenticator "cassandra" "cassandra") + let auth = Nothing + pool <- newPool [("localhost", "9042")] "test" auth -- servers, keyspace, auth runCas pool $ do ignoreDropFailure $ liftIO . print =<< executeSchema QUORUM dropLists () liftIO . print =<< executeSchema QUORUM createLists () @@ -39,4 +42,3 @@ executeWrite QUORUM insert (u1, t) liftIO . print =<< executeRows QUORUM select () - diff --git a/tests/test-timeuuid.hs b/tests/test-timeuuid.hs --- a/tests/test-timeuuid.hs +++ b/tests/test-timeuuid.hs @@ -32,10 +32,13 @@ ignoreDropFailure :: Cas () -> Cas () ignoreDropFailure code = code `catch` \exc -> case exc of ConfigError _ _ -> return () -- Ignore the error if the table doesn't exist + Invalid _ _ -> return () _ -> throw exc main = do - pool <- newPool [("localhost", "9042")] "test" -- servers, keyspace + --let auth = Just (PasswordAuthenticator "cassandra" "cassandra") + let auth = Nothing + pool <- newPool [("localhost", "9042")] "test" auth -- servers, keyspace, auth runCas pool $ do ignoreDropFailure $ liftIO . print =<< executeSchema QUORUM dropLists () liftIO . print =<< executeSchema QUORUM createLists () @@ -47,4 +50,3 @@ executeWrite QUORUM insert (u2, t) liftIO . print =<< executeRows QUORUM select () - diff --git a/tests/test-varint.hs b/tests/test-varint.hs --- a/tests/test-varint.hs +++ b/tests/test-varint.hs @@ -26,10 +26,13 @@ ignoreDropFailure :: Cas () -> Cas () ignoreDropFailure code = code `catch` \exc -> case exc of ConfigError _ _ -> return () -- Ignore the error if the table doesn't exist + Invalid _ _ -> return () _ -> throw exc main = do - pool <- newPool [("localhost", "9042")] "test" -- servers, keyspace + --let auth = Just (PasswordAuthenticator "cassandra" "cassandra") + let auth = Nothing + pool <- newPool [("localhost", "9042")] "test" auth -- servers, keyspace, auth runCas pool $ do ignoreDropFailure $ liftIO . print =<< executeSchema QUORUM dropLists () liftIO . print =<< executeSchema QUORUM createLists () @@ -54,8 +57,7 @@ executeWrite QUORUM insert (u8, -32769) executeWrite QUORUM insert (u9, -32768) executeWrite QUORUM insert (u10, -32767) - + let us = [u1,u2,u3,u4,u5,u6,u7,u8,u9,u10] forM_ us $ \u -> liftIO . print =<< executeRow QUORUM select u -