{-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE CPP #-} {-# LANGUAGE ViewPatterns #-} module XMPP ( module XMPPTypes , listenForXmppClients , listenForRemotePeers , seekRemotePeers , quitListening ) where import ServerC import XMPPTypes import SocketLike import ByteStringOperators import Data.HList import Network.Socket ( Family , connect , socketToHandle , sClose , Socket(..) , socket , SocketType(..) ) import Network.BSD ( PortNumber , getHostName , hostName , hostAliases , getProtocolNumber ) import System.IO ( BufferMode(..) , IOMode(..) , hSetBuffering ) import Control.Exception ( bracketOnError ) import Control.Concurrent.STM import Data.Conduit import qualified Data.Conduit.List as CL import qualified Data.Conduit.Binary as CB import Data.ByteString (ByteString) import qualified Data.ByteString.Char8 as S (pack,putStr,putStrLn,append) import qualified Data.ByteString.Lazy.Char8 as L ( putStrLn , fromChunks , unlines , hPutStrLn ) import Control.Concurrent (forkIO,killThread) import Control.Concurrent.Async import Control.Exception (handle,SomeException(..),finally) import Control.Monad.IO.Class import Control.Monad.Trans.Class import Control.Monad.Trans.Maybe import Todo import Control.Monad as Monad import Text.XML.Stream.Parse (parseBytes,content) import Text.XML.Stream.Render import Data.XML.Types as XML import Data.Text.Encoding as S (decodeUtf8,encodeUtf8) import Data.Text.Lazy.Encoding as L (decodeUtf8,encodeUtf8) import Data.Text.Lazy (toStrict) import GetHostByAddr import Data.Monoid import qualified Data.Sequence as Seq import Data.Foldable (toList) #ifdef RENDERFLUSH import Data.Conduit.Blaze #endif import Data.List (find) import qualified Text.Show.ByteString as L import NestingXML import qualified Data.Set as Set import qualified Data.Map as Map import GHC.Conc ( threadStatus , ThreadStatus(..) ) data Commands = Send [XML.Event] | QuitThread deriving Prelude.Show getNamesForPeer :: Peer -> IO [ByteString] getNamesForPeer LocalHost = fmap ((:[]) . S.pack) getHostName getNamesForPeer peer@(RemotePeer addr) = do ent <- getHostByAddr addr -- AF_UNSPEC addr let names = hostName ent : hostAliases ent return . map S.pack $ names xmlifyPresenceForClient :: Presence -> IO [XML.Event] xmlifyPresenceForClient (Presence jid stat) = do let n = name jid rsc = resource jid names <- getNamesForPeer (peer jid) let tostr p = L.decodeUtf8 $ n <$++> "@" L.fromChunks [p] <++?> "/" <++$> rsc jidstrs = fmap (toStrict . tostr) names return (concatMap presenceEvents jidstrs) where presenceEvents jidstr = [ EventBeginElement "presence" (("from",[ContentText jidstr]):typ stat) , EventBeginElement "show" [] , EventContent (ContentText . shw $ stat) , EventEndElement "show" , EventEndElement "presence" ] typ Offline = [("type",[ContentText "unavailable"])] typ _ = [] shw Available = "chat" shw Away = "away" shw Offline = "away" -- Is this right? prefix ## name = Name name Nothing (Just prefix) streamP name = Name name (Just "http://etherx.jabber.org/streams") (Just "stream") greet host = [ EventBeginDocument , EventBeginElement (streamP "stream") [("from",[ContentText host]) ,("id",[ContentText "someid"]) ,("xmlns",[ContentText "jabber:client"]) ,("xmlns:stream",[ContentText "http://etherx.jabber.org/streams"]) ,("version",[ContentText "1.0"]) ] , EventBeginElement (streamP "features") [] , EventBeginElement "bind" [("xmlns",[ContentText "urn:ietf:params:xml:ns:xmpp-bind"])] , EventEndElement "bind" {- -- , " " , " " -- , " DIGEST-MD5" , " PLAIN" , " " -} , EventEndElement (streamP "features") ] -- type Consumer i m r = forall o. ConduitM i o m r mawait :: Monad m => MaybeT (ConduitM i o m) i mawait = MaybeT await elementAttrs expected (EventBeginElement name attrs) | nameLocalName name==expected = return attrs elementAttrs _ _ = mzero eventIsBeginElement (EventBeginElement _ _) = True eventIsBeginElement _ = False eventIsEndElement (EventEndElement _) = True eventIsEndElement _ = False filterMapElement:: (Monad m, MonadPlus mp) => (Event -> mp a) -> Event -> mp a -> MaybeT (ConduitM Event o m) (mp a) filterMapElement ret opentag empty = loop (empty `mplus` ret opentag) 1 where loop ts 0 = return ts loop ts cnt = do tag <- mawait let ts' = mplus ts (ret tag) case () of _ | eventIsEndElement tag -> loop ts' (cnt-1) _ | eventIsBeginElement tag -> loop ts' (cnt+1) _ -> loop ts' cnt gatherElement :: (Monad m, MonadPlus mp) => Event -> mp Event -> MaybeT (ConduitM Event o m) (mp Event) gatherElement opentag empty = loop (empty `mplus` return opentag) 1 where loop ts 0 = return ts loop ts cnt = do tag <- mawait let ts' = mplus ts (return tag) case () of _ | eventIsEndElement tag -> loop ts' (cnt-1) _ | eventIsBeginElement tag -> loop ts' (cnt+1) _ -> loop ts' cnt {- sourceStanza :: Monad m => Event -> ConduitM Event Event m () sourceStanza opentag = yield opentag >> loop 1 where loop 0 = return () loop cnt = do e <- await let go tag cnt = yield tag >> loop cnt case e of Just tag | eventIsEndElement tag -> go tag (cnt-1) Just tag | eventIsBeginElement tag -> go tag (cnt+1) Just tag -> go tag cnt Nothing -> return () -} voidMaybeT body = (>> return ()) . runMaybeT $ body fixMaybeT f = (>> return ()) . runMaybeT . fix $ f iq_bind_reply id jid = [ EventBeginElement "{jabber:client}iq" [("type",[ContentText "result"]),("id",[ContentText id])] , EventBeginElement "{urn:ietf:params:xml:ns:xmpp-bind}bind" [("xmlns",[ContentText "urn:ietf:params:xml:ns:xmpp-bind"])] , EventBeginElement "{urn:ietf:params:xml:ns:xmpp-bind}jid" [] , EventContent (ContentText jid) , EventEndElement "{urn:ietf:params:xml:ns:xmpp-bind}jid" , EventEndElement "{urn:ietf:params:xml:ns:xmpp-bind}bind" , EventEndElement "{jabber:client}iq" ] uncontent cs = head $ map getText cs where getText (ContentText x) = x getText (ContentEntity x ) = x -- doIQ :: MonadIO m => Event -> MaybeT (ConduitM Event o m) () doIQ session cmdChan tag@(EventBeginElement name attrs) = do (_,uncontent->iq_id) <- MaybeT . return $ find (\(n,v)->isId n) attrs -- The 'id' attribute is REQUIRED for IQ stanzas. -- todo: handle it's absence more gracefully case (find (\(n,v)->isType n) attrs) of Just (_,[ContentText "get"]) -> discard Just (_,[ContentText "set"]) -> do fix $ \iqsetloop -> do setwhat <- mawait liftIO (putStrLn $ "IQ-set " ++ show setwhat) case setwhat of bind@(EventBeginElement name attrs) | isBind name -> do fix $ \again -> do rscElem <- mawait liftIO (putStrLn $ "IQ-set-bind " ++ show rscElem) case rscElem of bindchild@(EventBeginElement name _) | isResource name -> do let isContent (EventContent (ContentText v)) = return v isContent _ = mzero xs <- filterMapElement isContent bindchild Nothing case xs of Just rsrc -> liftIO $ do setResource session (L.fromChunks [S.encodeUtf8 rsrc]) jid <- getJID session atomically $ writeTChan cmdChan (Send $ iq_bind_reply iq_id (toStrict $ L.decodeUtf8 $ L.show jid) ) Nothing -> return () -- TODO: empty resource tag? void $ gatherElement bind Nothing bindchild@(EventBeginElement _ _) -> do liftIO (putStrLn "unknown bind child") gatherElement bindchild Nothing void $ gatherElement bind Nothing EventEndElement _ -> do liftIO (putStrLn "empty bind") -- TODO -- A server that supports resource binding MUST be able to -- generate a resource identifier on behalf of a client. A -- resource identifier generated by the server MUST be unique -- for that . _ -> again discard req@(EventBeginElement name attrs) -> do liftIO (putStrLn $ "IQ-set-unknown " ++ show req) gatherElement req Nothing discard endtag@(EventEndElement _) -> do liftIO (putStrLn $ "IQ-set-empty" ++ show endtag) _ -> iqsetloop Just (_,[ContentText "result"]) -> discard Just (_,[ContentText "error"]) -> discard Just _ -> discard -- error: type must be one of {get,set,result,error} Nothing -> discard -- error: The 'type' attribute is REQUIRED for IQ stanzas. where isId n = n=="id" isType n = n=="type" isResource n = n=="{urn:ietf:params:xml:ns:xmpp-bind}resource" isBind n = n=="{urn:ietf:params:xml:ns:xmpp-bind}bind" discard = do xs <- gatherElement tag Seq.empty prettyPrint "client-in: ignoring iq:" (toList xs) withJust (Just x) f = f x withJust Nothing f = return () whenJust acn f = do x <- acn withJust x f tagAttrs (EventBeginElement _ xs) = xs tagAttrs _ = [] tagName (EventBeginElement n _) = n tagName _ = "" handleIQSetBind session cmdChan stanza_id = do whenJust nextElement $ \child -> do let unhandledBind = liftIO $ putStrLn $ "unhandled-bind: "++show child case tagName child of "{urn:ietf:params:xml:ns:xmpp-bind}resource" -> do rsc <- lift content liftIO $ do putStrLn $ "iq-set-bind-resource " ++ show rsc setResource session (L.fromChunks [S.encodeUtf8 rsc]) jid <- getJID session atomically $ writeTChan cmdChan (Send $ iq_bind_reply stanza_id (toStrict $ L.decodeUtf8 $ L.show jid) ) _ -> unhandledBind iq_session_reply host stanza_id = [ EventBeginElement "{jabber:client}iq" [("id",[ContentText stanza_id]) ,("from",[ContentText host]) ,("type",[ContentText "result"]) ] , EventEndElement "{jabber:client}iq" ] handleIQSetSession session cmdChan stanza_id = do host <- liftIO $ do jid <- getJID session names <- getNamesForPeer (peer jid) return (S.decodeUtf8 . head $ names) liftIO . atomically . writeTChan cmdChan . Send $ iq_session_reply host stanza_id handleIQSet session cmdChan tag = do withJust (lookupAttrib "id" (tagAttrs tag)) $ \stanza_id -> do whenJust nextElement $ \child -> do let unhandledSet = liftIO $ putStrLn ("iq-set: "++show (stanza_id,child)) case tagName child of "{urn:ietf:params:xml:ns:xmpp-bind}bind" -> handleIQSetBind session cmdChan stanza_id "{urn:ietf:params:xml:ns:xmpp-session}session" -> handleIQSetSession session cmdChan stanza_id _ -> unhandledSet matchAttrib name value attrs = case find ( (==name) . fst) attrs of Just (_,[ContentText x]) | x==value -> True Just (_,[ContentEntity x]) | x==value -> True _ -> False lookupAttrib name attrs = case find ( (==name) . fst) attrs of Just (_,[ContentText x]) -> Just x Just (_,[ContentEntity x]) -> Just x _ -> Nothing iqTypeSet = "set" iqTypeGet = "get" iqTypeResult = "result" iqTypeError = "error" isIQOf (EventBeginElement name attrs) testType | name=="{jabber:client}iq" && matchAttrib "type" testType attrs = True isIQOf _ _ = False iq_service_unavailable host iq_id mjid req = [ EventBeginElement "{jabber:client}iq" [("type",[ContentText "error"]) ,("id",[ContentText iq_id]) -- , TODO: set "from" if isJust mjid ] , EventBeginElement req [] , EventEndElement req , EventBeginElement "{jabber:client}error" [("type",[ContentText "cancel"])] , EventBeginElement "{urn:ietf:params:xml:ns:xmpp-stanzas}service-unavailable" [] , EventEndElement "{urn:ietf:params:xml:ns:xmpp-stanzas}service-unavailable" , EventEndElement "{jabber:client}error" , EventEndElement "{jabber:client}iq" ] handleIQGet session cmdChan tag = do withJust (lookupAttrib "id" (tagAttrs tag)) $ \stanza_id -> do whenJust nextElement $ \child -> do host <- liftIO $ do jid <- getJID session names <- getNamesForPeer (peer jid) return (S.decodeUtf8 . head $ names) let unhandledGet req = do liftIO $ putStrLn ("iq-get: "++show (stanza_id,child)) liftIO . atomically . writeTChan cmdChan . Send $ iq_service_unavailable host stanza_id Nothing req case tagName child of -- "{http://jabber.org/protocol/disco#items}query" -> liftIO $ putStrLn "iq-get-query-items" -- "{urn:xmpp:ping}ping" -> todo req -> unhandledGet req fromClient :: (MonadThrow m,MonadIO m, XMPPSession session) => session -> TChan Commands -> Sink XML.Event m () fromClient session cmdChan = doNestingXML $ do let log = liftIO . L.putStrLn . ("(C) " <++>) send = liftIO . atomically . writeTChan cmdChan . Send withXML $ \begindoc -> do when (begindoc==EventBeginDocument) $ do log "begin-doc" withXML $ \xml -> do withJust (elementAttrs "stream" xml) $ \stream_attrs -> do log $ "stream atributes: " <++> bshow stream_attrs host <- liftIO $ do jid <- getJID session names <- getNamesForPeer (peer jid) return (S.decodeUtf8 . head $ names) send $ greet host fix $ \loop -> do log "waiting for stanza." whenJust nextElement $ \stanza -> do stanza_lvl <- nesting let unhandledStanza = do mb <- lift . runMaybeT $ gatherElement stanza Seq.empty withJust mb $ \xs -> prettyPrint "C: " (toList xs) case () of _ | stanza `isIQOf` iqTypeSet -> handleIQSet session cmdChan stanza _ | stanza `isIQOf` iqTypeGet -> handleIQGet session cmdChan stanza _ | otherwise -> unhandledStanza awaitCloser stanza_lvl loop log $ "end of stream" withXML $ \xml -> do log $ "end-of-document: " <++> bshow xml prettyPrint prefix xs = liftIO $ do CL.sourceList xs $= renderBytes (def { rsPretty=True }) =$= CB.lines $$ CL.mapM_ (S.putStrLn . (prefix `S.append`)) toClient :: MonadIO m => TChan Presence -> TChan Commands -> Source m [XML.Event] toClient pchan cmdChan = fix $ \loop -> do event <- liftIO . atomically $ orElse (fmap Left $ readTChan pchan) (fmap Right $ readTChan cmdChan) case event of Right QuitThread -> return () Right (Send xs) -> yield xs >> prettyPrint ">C: " xs >> loop Left presence -> do xs <- liftIO $ xmlifyPresenceForClient presence yield xs loop handleClient :: (SocketLike sock, HHead l (XMPPClass session), XMPPSession session) => HCons sock (HCons t l) -> Source IO ByteString -> Sink ByteString IO () -> IO () handleClient st src snk = do let HCons sock (HCons _ st') = st session_factory = hHead st' pname <- getPeerName sock session <- newSession session_factory sock Prelude.putStrLn $ "PEER NAME: "++Prelude.show pname pchan <- subscribe session Nothing cmdChan <- atomically newTChan #ifdef RENDERFLUSH writer <- async ( toClient pchan cmdChan $$ flushList =$= renderBuilderFlush def =$= builderToByteStringFlush =$= discardFlush =$ snk ) #else writer <- async ( toClient pchan cmdChan $$ renderChunks =$ snk ) #endif finally ( src $= parseBytes def $$ fromClient session cmdChan ) $ do atomically $ writeTChan cmdChan QuitThread wait writer closeSession session listenForXmppClients :: (HList l, HHead l (XMPPClass session), HExtend e1 l2 l1, HExtend e l1 (HCons PortNumber l), XMPPSession session) => Family -> e1 -> e -> l2 -> IO ServerHandle listenForXmppClients addr_family session_factory port st = do doServer (addr_family .*. port .*. session_factory .*. st) handleClient #ifdef RENDERFLUSH flushList :: Monad m => ConduitM [a] (Flush a) m () flushList = fixMaybeT $ \loop -> do xs <- mawait lift ( CL.sourceList xs $$ CL.mapM_ (yield . Chunk) ) lift ( yield Flush ) loop discardFlush :: Monad m => ConduitM (Flush a) a m () discardFlush = fixMaybeT $ \loop -> do x <- mawait let unchunk (Chunk a) = a ischunk (Chunk _) = True ischunk _ = False lift . when (ischunk x) $ yield (unchunk x) loop #else renderChunks :: (MonadUnsafeIO m, MonadIO m) => ConduitM [Event] ByteString m () renderChunks = fixMaybeT $ \loop -> do xs <- mawait lift . when (not . null $ xs) $ ( CL.sourceList xs $= renderBytes def $$ CL.mapM_ yield ) loop #endif listenForRemotePeers :: (HList l, HHead l (XMPPClass session), HExtend e l1 (HCons PortNumber l), HExtend e1 l2 l1, XMPPSession session) => Family -> e1 -> e -> l2 -> IO ServerHandle listenForRemotePeers addrfamily session_factory port st = do doServer (addrfamily .*. port .*. session_factory .*. st) handlePeer handlePeer :: (SocketLike sock, HHead l (XMPPClass session), XMPPSession session) => HCons sock (HCons t1 l) -> Source IO ByteString -> t -> IO () handlePeer st src snk = do let HCons sock (HCons _ st') = st session_factory = hHead st' name <- fmap bshow $ getPeerName sock L.putStrLn $ "(P) connected " <++> name jids <- newTVarIO Set.empty session <- newSession session_factory sock finally ( src $= parseBytes def $$ fromPeer session ) $ do L.putStrLn $ "(P) disconnected " <++> name js <- fmap Set.toList (readTVarIO jids) let offline jid = Presence jid Offline forM_ js $ announcePresence session . offline closeSession session fromPeer :: (MonadThrow m,MonadIO m, XMPPSession session) => session -> Sink XML.Event m () fromPeer session = doNestingXML $ do let log = liftIO . L.putStrLn . ("(P) " <++>) withXML $ \begindoc -> do when (begindoc==EventBeginDocument) $ do log "begin-doc" withXML $ \xml -> do withJust (elementAttrs "stream" xml) $ \stream_attrs -> do log $ "stream atributes: " <++> bshow stream_attrs fix $ \loop -> do log "waiting for stanza." whenJust nextElement $ \stanza -> do stanza_lvl <- nesting awaitCloser stanza_lvl loop log $ "end of stream" withXML $ \xml -> do log $ "end-of-document: " <++> bshow xml {- seekRemotePeers :: XMPPConfig config => config -> TChan Presence -> IO () seekRemotePeers config chan = do putStrLn "unimplemented: seekRemotePeers" -- TODO return () -} data OutBoundMessage = OutBoundPresence Presence deriving Prelude.Show newServerConnections = atomically $ newTVar Map.empty connect_to_server chan peer = (>> return ()) . runMaybeT $ do let port = 5269 :: Int connected <- liftIO . async $ connect' (peerAddr peer) port -- We'll cache Presence notifications until the socket -- is ready. cached <- liftIO $ newIORef Map.empty sock <- MaybeT . fix $ \loop -> do e <- atomically $ orElse (fmap Right $ waitSTM connected) (fmap Left $ readTChan chan) case e of Left (OutBoundPresence (Presence jid Offline)) -> do cached_map <- readIORef cached writeIORef cached (Map.delete jid cached_map) loop Left (OutBoundPresence p@(Presence jid st)) -> do cached_map <- readIORef cached writeIORef cached (Map.insert jid st cached_map) loop {- Left event -> do L.putStrLn $ "REMOTE-OUT DISCARDED: " <++> bshow event loop -} Right sock -> return sock liftIO $ do h <- socketToHandle sock ReadWriteMode hSetBuffering h NoBuffering L.hPutStrLn h "" L.putStrLn $ "OUT peer: " cache <- fmap Map.assocs . readIORef $ cached writeIORef cached Map.empty -- hint garbage collector: we're done with this forM_ cache $ \(jid,st) -> do r <- xmlifyPresenceForPeer sock (Presence jid st) L.hPutStrLn h r L.putStrLn $ "OUT peer: (cache)\n" <++> r <++> "\n" fix $ \loop -> do event <- atomically $ readTChan chan case event of OutBoundPresence p -> do r <- xmlifyPresenceForPeer sock p L.hPutStrLn h r L.putStrLn $ "OUT peer:\n" <++> r <++> "\n" loop L.hPutStrLn h "" L.putStrLn $ "OUT peer: " connect' :: SockAddr -> Int -> IO (Maybe Socket) connect' addr port = do proto <- getProtocolNumber "tcp" {- -- Given (host :: HostName) ... let hints = defaultHints { addrFlags = [AI_ADDRCONFIG] , addrProtocol = proto , addrSocketType = Stream } addrs <- getAddrInfo (Just hints) (Just host) (Just serv) firstSuccessful $ map tryToConnect addrs -} let getport (SockAddrInet port _) = port getport (SockAddrInet6 port _ _ _) = port let withPort (SockAddrInet _ a) port = SockAddrInet (toEnum port) a withPort (SockAddrInet6 _ a b c) port = SockAddrInet6 (toEnum port) a b c let doException (SomeException e) = do L.putStrLn $ "\nFailed to reach "<++> showPeer (RemotePeer addr) <++> " on port "<++>bshow port<++>": " <++> bshow e return Nothing handle doException $ tryToConnect proto (addr `withPort` port) where tryToConnect proto addr = bracketOnError (socket (socketFamily addr) Stream proto) (sClose ) -- only done if there's an error (\sock -> do connect sock addr return (Just sock) -- socketToHandle sock ReadWriteMode ) sendMessage cons msg peer = do found <- atomically $ do consmap <- readTVar cons return (Map.lookup peer consmap) let newEntry = do chan <- atomically newTChan t <- forkIO $ connect_to_server chan peer -- L.putStrLn $ "remote-map new: " <++> showPeer peer return (True,(chan,t)) (is_new,entry) <- maybe newEntry ( \(chan,t) -> do st <- threadStatus t let running = do -- L.putStrLn $ "remote-map, thread running: " <++> showPeer peer return (False,(chan,t)) died = do -- L.putStrLn $ "remote-map, thread died("<++>bshow st<++>"): " <++> showPeer peer newEntry case st of ThreadRunning -> running ThreadBlocked _ -> running ThreadDied -> died ThreadFinished -> died ) found -- L.putStrLn $ "sendMessage ->"<++>showPeer peer<++>": "<++>bshow msg atomically $ writeTChan (fst entry) msg when is_new . atomically $ readTVar cons >>= writeTVar cons . Map.insert peer entry seekRemotePeers :: XMPPConfig config => config -> TChan Presence -> IO b0 seekRemotePeers config chan = do server_connections <- newServerConnections fix $ \loop -> do event <- atomically $ readTChan chan case event of p@(Presence jid stat) | not (is_remote (peer jid)) -> do -- L.putStrLn $ "seekRemotePeers: " <++> L.show jid <++> " " <++> bshow stat runMaybeT $ do u <- MaybeT . return $ name jid subscribers <- liftIO $ do subs <- getSubscribers config u mapM parseHostNameJID subs -- liftIO . L.putStrLn $ "subscribers: " <++> bshow subscribers let peers = Set.map peer (Set.fromList subscribers) forM_ (Set.toList peers) $ \peer -> do when (is_remote peer) $ liftIO $ sendMessage server_connections (OutBoundPresence p) peer -- TODO: send presence probes for buddies -- TODO: cache remote presences for clients _ -> return (Just ()) loop xmlifyPresenceForPeer sock (Presence jid stat) = do -- TODO: accept socket argument and determine local ip address -- connected to this peer. addr <- getSocketName sock let n = name jid rsc = resource jid jid_str = n <$++> "@" showPeer (RemotePeer addr) <++?> "/" <++$> rsc return . L.unlines $ [ " typ stat <++> ">" , "" <++> shw stat <++> "" , "" ] where typ Offline = " type='unavailable'" typ _ = "" shw Available = "chat" shw Away = "away" shw Offline = "away" -- Is this right?