ige.ClientMngr.ClientMngr.resetAIAccounts() - Code Metrics - ospaceteam/outerspace - Measure and Improve Code Quality continuously with Scrutinizer

ige.ClientMngr.ClientMngr.resetAIAccounts() A
last analyzed 2021-09-11 21:46 UTC

↳ Parent: ige.ClientMngr

Complexity

Conditions

Size

Total Lines	7
Code Lines	7

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	3
eloc	7
nop	1
dl	0
loc	7
rs	10
c	0
b	0
f	0

#
#  Copyright 2001 - 2016 Ludek Smid [http://www.ospace.net/]
#
#  This file is part of Outer Space.
#
#  Outer Space is free software; you can redistribute it and/or modify
#  it under the terms of the GNU General Public License as published by
#  the Free Software Foundation; either version 2 of the License, or
#  (at your option) any later version.
#
#  Outer Space is distributed in the hope that it will be useful,
#  but WITHOUT ANY WARRANTY; without even the implied warranty of
#  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
#  GNU General Public License for more details.
#
#  You should have received a copy of the GNU General Public License
#  along with Outer Space; if not, write to the Free Software
#  Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA
#

import os
import hashlib
import random
import time
import log
import ige
from ige import SecurityException
from ige.Const import ADMIN_LOGIN
import Authentication
from account import Account, AIAccount, AdminAccount, passwordGen
from ai_parser import AIList
from IDataHolder import IDataHolder

class ClientMngr:

    def __init__(self, database, authMethod, configDir):
        self.configDir = configDir
        self.authMethod = authMethod
        if not self.authMethod:
            self.authMethod = Authentication.defaultMethod
        if ige.igeRuntimeMode == 1:
            Authentication.init(self.authMethod, 2048)
        elif ige.igeRuntimeMode == 0:
            # it is minimum to cater for AI generated passwords
            Authentication.init(self.authMethod, 512)
        self._filename = os.path.join(self.configDir, 'accounts')
        self.sessions = {}
        #
        self.accounts = database
        self._initAdminAccount()
        self.generateAIList()

    def shutdown(self):
        self.accounts.shutdown()

    def checkpoint(self):
        self.accounts.checkpoint()

    def backup(self, basename):
        self.accounts.backup(basename)

    def exists(self, login):
        return self.accounts.has_key(str(login))

    def __getitem__(self, login):
        return self.accounts[str(login)]

    def _initAdminAccount(self):
        # create special key

        if self.accounts.has_key(ADMIN_LOGIN):
            self.accounts[ADMIN_LOGIN].passwdHashed = False # Needs plaintext login from token
            password = passwordGen()
            self.accounts[ADMIN_LOGIN].setPassword(password)
        else:
            log.message("No administator account found! (looking for '%s')" % ADMIN_LOGIN)
            log.message("Creating default account")
            # create account
            account = AdminAccount()
            # update
            password = account.passwd
            self.accounts.create(account, id = str(account.login))
        with open(os.path.join(self.configDir, "token"), "w") as tokenFile:
            tokenFile.write(password)

    def createAccount(self, sid, login, safePasswd, nick, email):
        log.message('Creating account', login, nick, email)
        session = self.getSession(sid)
        plainPassword = Authentication.unwrapUserPassword(safePasswd, session.challenge)
        login = login.strip()
        plainPassword = plainPassword.strip()
        nick = nick.strip()
        # check requirement
        if len(login) < ige.Const.ACCOUNT_LOGIN_MIN_LEN:
            raise SecurityException('Login is too short.')
        if len(plainPassword) < ige.Const.ACCOUNT_PASSWD_MIN_LEN:
            raise SecurityException('Password is too short.')
        if len(nick) < ige.Const.ACCOUNT_NICK_MIN_LEN:
            raise SecurityException('Nick is too short.')
        # check login, nick and uid
        for key in self.accounts.keys():
            account = self.accounts[key]
            if account.login == login:
                raise SecurityException('Login already used.')
            elif account.nick == nick:
                raise SecurityException('Nick already used.')
            elif account.email == email:
                raise SecurityException('E-mail already used.')
        # create account
        account = Account(login, nick, email, plainPassword)
        # update
        self.accounts.create(account, id = str(account.login))
        log.message('Account created, confirmation token:', account.confToken)
        # TODO send confirmation token to the email address
        return 1, None

    def createAIAccount(self, login, nick, aiType):
        if self.accounts.has_key(login):
            log.message('AI account already exists, no work needed.', login, nick)
            password = self.accounts[login].passwd
            return password, None

        log.message('Creating AI account', login, nick)
        login = login.strip()
        nick = nick.strip()
        # create account
        account = AIAccount(login, nick, aiType)
        # update
        self.accounts.create(account, id = str(account.login))
        log.message('AI account created')
        self.generateAIList()
        return 1, None

    def generateAIList(self):
        aiList = AIList(self.configDir)
        aiList.removeAll()
        for login in self.accounts.keys():
            account = self.accounts[login]
            if not account.isAI:
                continue
            aiList.add(login, account.passwd, account.aiType)
        aiList.save()
        log.message('AI list regenerated')

    def removeAIAccount(self, login):
        self.accounts.delete(login)
        self.generateAIList()
        return 1, None

    def resetAIAccounts(self):
        for login in self.accounts.keys():
            account = self.accounts[login]
            if account.isAI:
                self.accounts.delete(login)
        self.generateAIList()
        return 1, None

    def hello(self, sid, clientId):
        log.debug(clientId, 'connected. User', repr(clientId))
        # create sort of cookie
        while 1:
            sid = hashlib.sha256(str(random.random())).hexdigest()
            if not self.sessions.has_key(sid):
                break
        challenge = Authentication.getWelcomeString(self.authMethod)
        session = Session(sid)
        session.challenge = challenge
        session.clientIdent = clientId
        self.sessions[sid] = session
        return (sid, challenge), None

    def login(self, sid, login, safePassword, hostID):
        login = login.strip()
        if not login:
            raise SecurityException("Specify login, please.")
        if login in ():
            raise SecurityException("Account blocked")
        log.debug(sid, 'login', repr(login), 'hostid', hostID)
        login = str(login)
        challenge = self.sessions[sid].challenge
        log.debug("Trying local login for user", login)
        if not self.accounts.has_key(login):
            raise SecurityException('Wrong login and/or password.')
        account = self.accounts[login]
        plainPassword = Authentication.unwrapUserPassword(safePassword, challenge)
        if not account.verifyPassword(plainPassword):
            raise SecurityException('Wrong login and/or password.')
        # setup session
        self.sessions[sid].setAttrs(account.login, account.nick, account.email)
        account.lastLogin = time.time()
        account.addHostID(hostID)
        return 1, None

    def getAccountData(self, sid):
        session = self.getSession(sid)
        result = IDataHolder()
        result.login = session.login
        result.nick = session.nick
        result.email = session.email
        return result, None

    def getSession(self, sid):
        session = self.sessions.get(sid, None)
        if not session:
            raise SecurityException('No such session id.')
        return session

    def getSessionByCID(self, cid):
        # TODO more effective - used by GameMngr.sendMessage
        for sid in self.sessions.keys():
            session = self.sessions[sid]
            if session.cid == cid:
                return session
        return None

    def logout(self, sid):
        session = self.sessions.get(sid, None)
        if session:
            try:
                log.debug(sid, 'logout', self.sessions[sid].login)
            except AttributeError:
                pass
            del self.sessions[sid]
            return 1, None
        else:
            raise SecurityException('No such session id.')

    def changePassword(self, sid, safeOld, safeNew):
        session = self.sessions[sid]
        if not session:
            raise SecurityException('No such session id.')
        challenge = session.challenge
        account = self.accounts[session.login]
        oldPassword = Authentication.unwrapUserPassword(safeOld, challenge)
        newPassword = Authentication.unwrapUserPassword(safeNew, challenge)
        if not account.verifyPassword(oldPassword):
            raise SecurityException('Wrong login and/or password.')
        if len(newPassword) < ige.Const.ACCOUNT_PASSWD_MIN_LEN:
            raise SecurityException('Password is too short.')
        account.setPassword(newPassword)
        log.debug('Password of account {0} successfully changed.'.format(session.login))
        return None, None

    def cleanupSessions(self, sid):
        session = self.sessions.get(sid, None)
        if session:
            log.debug('cleaning up sessions')
            now = time.time()
            deleted = 0
            for id in self.sessions.keys():
                if self.sessions[id].timeout < now:
                    log.debug("Deleting session", self.sessions[id].sid, getattr(self.sessions[id], "login", "<unknown>"))
                    del self.sessions[id]
                    deleted += 1
            log.debug('cleanup finished (%d active, %d deleted)' % (len(self.sessions), deleted))
            return None, None
        else:
            raise SecurityException('No such session id.')

    def exportAccounts(self, sid):
        # check admin
        session = self.getSession(sid)
        if session.login != ADMIN_LOGIN:
            raise SecurityException('You cannot issue this command.')
        # export accounts
        f = open(os.path.join(self.configDir,"accounts.txt"), "w")
        for account in self.accounts.keys():
            account = self.accounts[account]
            print >>f, "%s\t%s\t%s\t%s" % (
                account.nick.encode("utf-8"),
                account.login.encode("utf-8"),
                account.passwd.encode("utf-8"),
                account.email.encode("utf-8")
            )
        f.close()
        return None, None

    def serverShutdown(self, sid):
        # check admin
        session = self.getSession(sid)
        if session.login != ADMIN_LOGIN:
            raise SecurityException('You cannot issue this command.')
        log.message('Shutting down server')
        import ige.RPCServer
        ige.RPCServer.running = 0
        return 1, None

    # new rpc interface wrappers
    def rpc_hello(self, login, clientId):
        return self.hello(None, login, clientId)[0]

    def rpc_login(self, sid, login, cpasswd, hostID):
        return self.login(sid, login, cpasswd, hostID)[0]

    def rpc_logout(self, sid):
        return self.logout(sid)[0]

    def rpc_createAccount(self, sid, login, passwd, nick, email):
        return self.createAccount(sid, login, passwd, nick, email)[0]

# Session class keeps various session data
class Session:

    def __init__(self, sid):
        self.cid = None
        self.sid = sid
        self.messages = {}
        self.touch()

    def setAttrs(self, login, nick, email):
        self.login = login
        self.nick = nick
        self.email = email

    def touch(self):
        # 10 minutes timeout of session
        self.timeout = time.time() + 10 * 60

class Mapping(dict):
    pass


1			#
2			# Copyright 2001 - 2016 Ludek Smid [http://www.ospace.net/]
3			#
4			# This file is part of Outer Space.
5			#
6			# Outer Space is free software; you can redistribute it and/or modify
7			# it under the terms of the GNU General Public License as published by
8			# the Free Software Foundation; either version 2 of the License, or
9			# (at your option) any later version.
10			#
11			# Outer Space is distributed in the hope that it will be useful,
12			# but WITHOUT ANY WARRANTY; without even the implied warranty of
13			# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14			# GNU General Public License for more details.
15			#
16			# You should have received a copy of the GNU General Public License
17			# along with Outer Space; if not, write to the Free Software
18			# Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
19			#
20
21			import os
22			import hashlib
23			import random
24			import time
25			import log
26			import ige
27			from ige import SecurityException
28			from ige.Const import ADMIN_LOGIN
29			import Authentication
30			from account import Account, AIAccount, AdminAccount, passwordGen
31			from ai_parser import AIList
32			from IDataHolder import IDataHolder
33
34			class ClientMngr:
35
36			def __init__(self, database, authMethod, configDir):
37			self.configDir = configDir
38			self.authMethod = authMethod
39			if not self.authMethod:
40			self.authMethod = Authentication.defaultMethod
41			if ige.igeRuntimeMode == 1:
42			Authentication.init(self.authMethod, 2048)
43			elif ige.igeRuntimeMode == 0:
44			# it is minimum to cater for AI generated passwords
45			Authentication.init(self.authMethod, 512)
46			self._filename = os.path.join(self.configDir, 'accounts')
47			self.sessions = {}
48			#
49			self.accounts = database
50			self._initAdminAccount()
51			self.generateAIList()
52
53			def shutdown(self):
54			self.accounts.shutdown()
55
56			def checkpoint(self):
57			self.accounts.checkpoint()
58
59			def backup(self, basename):
60			self.accounts.backup(basename)
61
62			def exists(self, login):
63			return self.accounts.has_key(str(login))
64
65			def __getitem__(self, login):
66			return self.accounts[str(login)]
67
68			def _initAdminAccount(self):
69			# create special key
70
71			if self.accounts.has_key(ADMIN_LOGIN):
72			self.accounts[ADMIN_LOGIN].passwdHashed = False # Needs plaintext login from token
73			password = passwordGen()
74			self.accounts[ADMIN_LOGIN].setPassword(password)
75			else:
76			log.message("No administator account found! (looking for '%s')" % ADMIN_LOGIN)
77			log.message("Creating default account")
78			# create account
79			account = AdminAccount()
80			# update
81			password = account.passwd
82			self.accounts.create(account, id = str(account.login))
83			with open(os.path.join(self.configDir, "token"), "w") as tokenFile:
84			tokenFile.write(password)
85
86			def createAccount(self, sid, login, safePasswd, nick, email):
87			log.message('Creating account', login, nick, email)
88			session = self.getSession(sid)
89			plainPassword = Authentication.unwrapUserPassword(safePasswd, session.challenge)
90			login = login.strip()
91			plainPassword = plainPassword.strip()
92			nick = nick.strip()
93			# check requirement
94			if len(login) < ige.Const.ACCOUNT_LOGIN_MIN_LEN:
95			raise SecurityException('Login is too short.')
96			if len(plainPassword) < ige.Const.ACCOUNT_PASSWD_MIN_LEN:
97			raise SecurityException('Password is too short.')
98			if len(nick) < ige.Const.ACCOUNT_NICK_MIN_LEN:
99			raise SecurityException('Nick is too short.')
100			# check login, nick and uid
101			for key in self.accounts.keys():
102			account = self.accounts[key]
103			if account.login == login:
104			raise SecurityException('Login already used.')
105			elif account.nick == nick:
106			raise SecurityException('Nick already used.')
107			elif account.email == email:
108			raise SecurityException('E-mail already used.')
109			# create account
110			account = Account(login, nick, email, plainPassword)
111			# update
112			self.accounts.create(account, id = str(account.login))
113			log.message('Account created, confirmation token:', account.confToken)
114			# TODO send confirmation token to the email address
115			return 1, None
116
117			def createAIAccount(self, login, nick, aiType):
118			if self.accounts.has_key(login):
119			log.message('AI account already exists, no work needed.', login, nick)
120			password = self.accounts[login].passwd
121			return password, None
122
123			log.message('Creating AI account', login, nick)
124			login = login.strip()
125			nick = nick.strip()
126			# create account
127			account = AIAccount(login, nick, aiType)
128			# update
129			self.accounts.create(account, id = str(account.login))
130			log.message('AI account created')
131			self.generateAIList()
132			return 1, None
133
134			def generateAIList(self):
135			aiList = AIList(self.configDir)
136			aiList.removeAll()
137			for login in self.accounts.keys():
138			account = self.accounts[login]
139			if not account.isAI:
140			continue
141			aiList.add(login, account.passwd, account.aiType)
142			aiList.save()
143			log.message('AI list regenerated')
144
145			def removeAIAccount(self, login):
146			self.accounts.delete(login)
147			self.generateAIList()
148			return 1, None
149
150			def resetAIAccounts(self):
151			for login in self.accounts.keys():
152			account = self.accounts[login]
153			if account.isAI:
154			self.accounts.delete(login)
155			self.generateAIList()
156			return 1, None
157
158			def hello(self, sid, clientId):
159			log.debug(clientId, 'connected. User', repr(clientId))
160			# create sort of cookie
161			while 1:
162			sid = hashlib.sha256(str(random.random())).hexdigest()
163			if not self.sessions.has_key(sid):
164			break
165			challenge = Authentication.getWelcomeString(self.authMethod)
166			session = Session(sid)
167			session.challenge = challenge
168			session.clientIdent = clientId
169			self.sessions[sid] = session
170			return (sid, challenge), None
171
172			def login(self, sid, login, safePassword, hostID):
173			login = login.strip()
174			if not login:
175			raise SecurityException("Specify login, please.")
176			if login in ():
177			raise SecurityException("Account blocked")
178			log.debug(sid, 'login', repr(login), 'hostid', hostID)
179			login = str(login)
180			challenge = self.sessions[sid].challenge
181			log.debug("Trying local login for user", login)
182			if not self.accounts.has_key(login):
183			raise SecurityException('Wrong login and/or password.')
184			account = self.accounts[login]
185			plainPassword = Authentication.unwrapUserPassword(safePassword, challenge)
186			if not account.verifyPassword(plainPassword):
187			raise SecurityException('Wrong login and/or password.')
188			# setup session
189			self.sessions[sid].setAttrs(account.login, account.nick, account.email)
190			account.lastLogin = time.time()
191			account.addHostID(hostID)
192			return 1, None
193
194			def getAccountData(self, sid):
195			session = self.getSession(sid)
196			result = IDataHolder()
197			result.login = session.login
198			result.nick = session.nick
199			result.email = session.email
200			return result, None
201
202			def getSession(self, sid):
203			session = self.sessions.get(sid, None)
204			if not session:
205			raise SecurityException('No such session id.')
206			return session
207
208			def getSessionByCID(self, cid):
209			# TODO more effective - used by GameMngr.sendMessage
210			for sid in self.sessions.keys():
211			session = self.sessions[sid]
212			if session.cid == cid:
213			return session
214			return None
215
216			def logout(self, sid):
217			session = self.sessions.get(sid, None)
218			if session:
219			try:
220			log.debug(sid, 'logout', self.sessions[sid].login)
221			except AttributeError:
222			pass
223			del self.sessions[sid]
224			return 1, None
225			else:
226			raise SecurityException('No such session id.')
227
228			def changePassword(self, sid, safeOld, safeNew):
229			session = self.sessions[sid]
230			if not session:
231			raise SecurityException('No such session id.')
232			challenge = session.challenge
233			account = self.accounts[session.login]
234			oldPassword = Authentication.unwrapUserPassword(safeOld, challenge)
235			newPassword = Authentication.unwrapUserPassword(safeNew, challenge)
236			if not account.verifyPassword(oldPassword):
237			raise SecurityException('Wrong login and/or password.')
238			if len(newPassword) < ige.Const.ACCOUNT_PASSWD_MIN_LEN:
239			raise SecurityException('Password is too short.')
240			account.setPassword(newPassword)
241			log.debug('Password of account {0} successfully changed.'.format(session.login))
242			return None, None
243
244			def cleanupSessions(self, sid):
245			session = self.sessions.get(sid, None)
246			if session:
247			log.debug('cleaning up sessions')
248			now = time.time()
249			deleted = 0
250			for id in self.sessions.keys():
251			if self.sessions[id].timeout < now:
252			log.debug("Deleting session", self.sessions[id].sid, getattr(self.sessions[id], "login", "<unknown>"))
253			del self.sessions[id]
254			deleted += 1
255			log.debug('cleanup finished (%d active, %d deleted)' % (len(self.sessions), deleted))
256			return None, None
257			else:
258			raise SecurityException('No such session id.')
259
260			def exportAccounts(self, sid):
261			# check admin
262			session = self.getSession(sid)
263			if session.login != ADMIN_LOGIN:
264			raise SecurityException('You cannot issue this command.')
265			# export accounts
266			f = open(os.path.join(self.configDir,"accounts.txt"), "w")
267			for account in self.accounts.keys():
268			account = self.accounts[account]
269			print >>f, "%s\t%s\t%s\t%s" % (
270			account.nick.encode("utf-8"),
271			account.login.encode("utf-8"),
272			account.passwd.encode("utf-8"),
273			account.email.encode("utf-8")
274			)
275			f.close()
276			return None, None
277
278			def serverShutdown(self, sid):
279			# check admin
280			session = self.getSession(sid)
281			if session.login != ADMIN_LOGIN:
282			raise SecurityException('You cannot issue this command.')
283			log.message('Shutting down server')
284			import ige.RPCServer
285			ige.RPCServer.running = 0
286			return 1, None
287
288			# new rpc interface wrappers
289			def rpc_hello(self, login, clientId):
290			return self.hello(None, login, clientId)[0]
291
292			def rpc_login(self, sid, login, cpasswd, hostID):
293			return self.login(sid, login, cpasswd, hostID)[0]
294
295			def rpc_logout(self, sid):
296			return self.logout(sid)[0]
297
298			def rpc_createAccount(self, sid, login, passwd, nick, email):
299			return self.createAccount(sid, login, passwd, nick, email)[0]
300
301			# Session class keeps various session data
302			class Session:
303
304			def __init__(self, sid):
305			self.cid = None
306			self.sid = sid
307			self.messages = {}
308			self.touch()
309
310			def setAttrs(self, login, nick, email):
311			self.login = login
312			self.nick = nick
313			self.email = email
314
315			def touch(self):
316			# 10 minutes timeout of session
317			self.timeout = time.time() + 10 * 60
318
319			class Mapping(dict):
320			pass
321

ospaceteam / outerspace

ige.ClientMngr.ClientMngr.resetAIAccounts() A last analyzed 2021-09-11 21:46 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

ige.ClientMngr.ClientMngr.resetAIAccounts() A
last analyzed 2021-09-11 21:46 UTC