Toggle Light / Dark / Auto color theme
Toggle table of contents sidebar
Source code for matridge.gateway
import logging
from typing import Optional
from nio.responses import LoginError
from slidge import BaseGateway , FormField , global_config
from slixmpp import JID
from . import config
from .matrix import AuthenticationClient
from .util import Credentials
[docs]
class Gateway ( BaseGateway ):
[docs]
REGISTRATION_FIELDS = [
FormField ( var = "homeserver" , label = "Home Server" , required = True ),
FormField ( var = "username" , label = "User name" , required = True ),
FormField ( var = "password" , label = "Password" , required = True , private = True ),
FormField (
var = "device" ,
label = "Device name" ,
value = f "matridge on { getattr ( global_config , 'JID' , 'dev' ) } " ,
required = True ,
),
]
[docs]
REGISTRATION_INSTRUCTIONS : str = "Enter your credentials"
[docs]
COMPONENT_NAME = "matridge"
"""Name of the component, as seen in service discovery by XMPP clients"""
[docs]
COMPONENT_TYPE = "matrix"
[docs]
COMPONENT_AVATAR = "https://upload.wikimedia.org/wikipedia/commons/thumb/7/7c/Matrix_icon.svg/209px-Matrix_icon.svg.png"
[docs]
ROSTER_GROUP : str = "matrix"
[docs]
MARK_ALL_MESSAGES = False
def __init__ ( self ):
super () . __init__ ()
if config . NIO_SILENT :
logging . getLogger ( "peewee" ) . setLevel ( logging . WARNING )
logging . getLogger ( "nio.responses" ) . setLevel ( logging . WARNING )
logging . getLogger ( "nio.rooms" ) . setLevel ( logging . WARNING )
[docs]
async def validate (
self , user_jid : JID , registration_form : dict [ str , Optional [ str ]]
) -> Credentials :
client = AuthenticationClient (
registration_form [ "homeserver" ], # type:ignore
registration_form [ "username" ], # type:ignore
user_jid ,
)
await client . fix_homeserver ()
resp = await client . login (
registration_form [ "password" ], # type:ignore
registration_form [ "device" ], # type:ignore
)
if isinstance ( resp , LoginError ):
log . debug ( "Failed login: %r " , resp )
raise PermissionError ( resp )
return client . get_credentials ( resp )
[docs]
async def unregister ( self , user ):
"""
Optionally override this if you need to clean additional
stuff after a user has been removed from the persistent user store.
By default, this just calls :meth:`BaseSession.logout`.
:param user:
"""
session = self . get_session_from_user ( user )
await session . matrix . logout () # type:ignore
[docs]
log = logging . getLogger ( __name__ )