ige.ClientMngr.ClientMngr.resetAIAccounts() - Code Metrics - Inspection of "Refactoring of planet production phase code." - ospaceteam/outerspace - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#192)

by Marek

created 2018-06-24 14:35 UTC

ige.ClientMngr.ClientMngr.resetAIAccounts() A

↳ 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
from ige import SecurityException
from ige.Const import ADMIN_LOGIN
import Authentication
from account import Account, AIAccount
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
        self._filename = os.path.join(self.configDir, 'accounts')
        self.sessions = {}
        #
        self.accounts = database
        # create special key
        if not self.accounts.has_key(ADMIN_LOGIN):
            log.message("No administator account found! (looking for '%s')" % ADMIN_LOGIN)
            log.message("Creating default account")
            self.createAccount(None, ADMIN_LOGIN, "tobechanged", "Administrator", "[email protected]")
        password = hashlib.sha1(str(random.randrange(0, 1e10))).hexdigest()
        open(os.path.join(self.configDir, "token"), "w").write(password)
        self.accounts[ADMIN_LOGIN].passwd = password
        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 createAccount(self, sid, login, passwd, nick, email):
        log.message('Creating account', login, nick, email)
        login = login.strip()
        passwd = passwd.strip()
        nick = nick.strip()
        # check requirement
        if len(login) < 4:
            raise SecurityException('Login is too short.')
        if len(passwd) < 4:
            raise SecurityException('Password is too short.')
        if len(nick) < 4:
            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()
        # update
        account.login = login
        account.passwd = passwd
        account.nick = nick
        account.email = email
        account.confToken = hashlib.md5('%s%s%d' % (login, email, time.time())).hexdigest()
        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()
        # update
        account.login = login
        account.nick = nick
        account.aiType = aiType
        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, login, clientId):
        log.debug(clientId, 'connected. User', repr(login))
        login = str(login)
        # 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
        account = self.accounts.get(login, None)
        return (sid, challenge), None

    def login(self, sid, login, cpasswd, 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]
        if not Authentication.verify(cpasswd, account.passwd, challenge):
            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, old, new):
        session = self.sessions[sid]
        if session:
            if self.accounts[session.login].passwd == old:
                self.accounts[session.login].passwd = new
                return 1, None
            else:
                raise SecurityException('Wrong password.')
        else:
            raise SecurityException('No such session id.')

    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			from ige import SecurityException
27			from ige.Const import ADMIN_LOGIN
28			import Authentication
29			from account import Account, AIAccount
30			from ai_parser import AIList
31			from IDataHolder import IDataHolder
32
33			class ClientMngr:
34
35			def __init__(self, database, authMethod, configDir):
36			self.configDir = configDir
37			self.authMethod = authMethod
38			if not self.authMethod:
39			self.authMethod = Authentication.defaultMethod
40			self._filename = os.path.join(self.configDir, 'accounts')
41			self.sessions = {}
42			#
43			self.accounts = database
44			# create special key
45			if not self.accounts.has_key(ADMIN_LOGIN):
46			log.message("No administator account found! (looking for '%s')" % ADMIN_LOGIN)
47			log.message("Creating default account")
48			self.createAccount(None, ADMIN_LOGIN, "tobechanged", "Administrator", "[email protected]")
49			password = hashlib.sha1(str(random.randrange(0, 1e10))).hexdigest()
50			open(os.path.join(self.configDir, "token"), "w").write(password)
51			self.accounts[ADMIN_LOGIN].passwd = password
52			self.generateAIList()
53
54			def shutdown(self):
55			self.accounts.shutdown()
56
57			def checkpoint(self):
58			self.accounts.checkpoint()
59
60			def backup(self, basename):
61			self.accounts.backup(basename)
62
63			def exists(self, login):
64			return self.accounts.has_key(str(login))
65
66			def __getitem__(self, login):
67			return self.accounts[str(login)]
68
69			def createAccount(self, sid, login, passwd, nick, email):
70			log.message('Creating account', login, nick, email)
71			login = login.strip()
72			passwd = passwd.strip()
73			nick = nick.strip()
74			# check requirement
75			if len(login) < 4:
76			raise SecurityException('Login is too short.')
77			if len(passwd) < 4:
78			raise SecurityException('Password is too short.')
79			if len(nick) < 4:
80			raise SecurityException('Nick is too short.')
81			# check login, nick and uid
82			for key in self.accounts.keys():
83			account = self.accounts[key]
84			if account.login == login:
85			raise SecurityException('Login already used.')
86			elif account.nick == nick:
87			raise SecurityException('Nick already used.')
88			elif account.email == email:
89			raise SecurityException('E-mail already used.')
90			# create account
91			account = Account()
92			# update
93			account.login = login
94			account.passwd = passwd
95			account.nick = nick
96			account.email = email
97			account.confToken = hashlib.md5('%s%s%d' % (login, email, time.time())).hexdigest()
98			self.accounts.create(account, id = str(account.login))
99			log.message('Account created, confirmation token:', account.confToken)
100			# TODO send confirmation token to the email address
101			return 1, None
102
103			def createAIAccount(self, login, nick, aiType):
104			if self.accounts.has_key(login):
105			log.message('AI account already exists, no work needed.', login, nick)
106			password = self.accounts[login].passwd
107			return password, None
108
109			log.message('Creating AI account', login, nick)
110			login = login.strip()
111			nick = nick.strip()
112			# create account
113			account = AIAccount()
114			# update
115			account.login = login
116			account.nick = nick
117			account.aiType = aiType
118			self.accounts.create(account, id = str(account.login))
119			log.message('AI account created')
120			self.generateAIList()
121			return 1, None
122
123			def generateAIList(self):
124			aiList = AIList(self.configDir)
125			aiList.removeAll()
126			for login in self.accounts.keys():
127			account = self.accounts[login]
128			if not account.isAI:
129			continue
130			aiList.add(login, account.passwd, account.aiType)
131			aiList.save()
132			log.message('AI list regenerated')
133
134			def removeAIAccount(self, login):
135			self.accounts.delete(login)
136			self.generateAIList()
137			return 1, None
138
139			def resetAIAccounts(self):
140			for login in self.accounts.keys():
141			account = self.accounts[login]
142			if account.isAI:
143			self.accounts.delete(login)
144			self.generateAIList()
145			return 1, None
146
147
148			def hello(self, sid, login, clientId):
149			log.debug(clientId, 'connected. User', repr(login))
150			login = str(login)
151			# create sort of cookie
152			while 1:
153			sid = hashlib.sha256(str(random.random())).hexdigest()
154			if not self.sessions.has_key(sid):
155			break
156			challenge = Authentication.getWelcomeString(self.authMethod)
157			session = Session(sid)
158			session.challenge = challenge
159			session.clientIdent = clientId
160			self.sessions[sid] = session
161			account = self.accounts.get(login, None)
162			return (sid, challenge), None
163
164			def login(self, sid, login, cpasswd, hostID):
165			login = login.strip()
166			if not login:
167			raise SecurityException("Specify login, please.")
168			if login in ():
169			raise SecurityException("Account blocked")
170			log.debug(sid, 'login', repr(login), 'hostid', hostID)
171			login = str(login)
172			challenge = self.sessions[sid].challenge
173			log.debug("Trying local login for user", login)
174			if not self.accounts.has_key(login):
175			raise SecurityException('Wrong login and/or password.')
176			account = self.accounts[login]
177			if not Authentication.verify(cpasswd, account.passwd, challenge):
178			raise SecurityException('Wrong login and/or password.')
179			# setup session
180			self.sessions[sid].setAttrs(account.login, account.nick, account.email)
181			account.lastLogin = time.time()
182			account.addHostID(hostID)
183			return 1, None
184
185			def getAccountData(self, sid):
186			session = self.getSession(sid)
187			result = IDataHolder()
188			result.login = session.login
189			result.nick = session.nick
190			result.email = session.email
191			return result, None
192
193			def getSession(self, sid):
194			session = self.sessions.get(sid, None)
195			if not session:
196			raise SecurityException('No such session id.')
197			return session
198
199			def getSessionByCID(self, cid):
200			# TODO more effective - used by GameMngr.sendMessage
201			for sid in self.sessions.keys():
202			session = self.sessions[sid]
203			if session.cid == cid:
204			return session
205			return None
206
207			def logout(self, sid):
208			session = self.sessions.get(sid, None)
209			if session:
210			try:
211			log.debug(sid, 'logout', self.sessions[sid].login)
212			except AttributeError:
213			pass
214			del self.sessions[sid]
215			return 1, None
216			else:
217			raise SecurityException('No such session id.')
218
219			def changePassword(self, sid, old, new):
220			session = self.sessions[sid]
221			if session:
222			if self.accounts[session.login].passwd == old:
223			self.accounts[session.login].passwd = new
224			return 1, None
225			else:
226			raise SecurityException('Wrong password.')
227			else:
228			raise SecurityException('No such session id.')
229
230			def cleanupSessions(self, sid):
231			session = self.sessions.get(sid, None)
232			if session:
233			log.debug('cleaning up sessions')
234			now = time.time()
235			deleted = 0
236			for id in self.sessions.keys():
237			if self.sessions[id].timeout < now:
238			log.debug("Deleting session", self.sessions[id].sid, getattr(self.sessions[id], "login", "<unknown>"))
239			del self.sessions[id]
240			deleted += 1
241			log.debug('cleanup finished (%d active, %d deleted)' % (len(self.sessions), deleted))
242			return None, None
243			else:
244			raise SecurityException('No such session id.')
245
246			def exportAccounts(self, sid):
247			# check admin
248			session = self.getSession(sid)
249			if session.login != ADMIN_LOGIN:
250			raise SecurityException('You cannot issue this command.')
251			# export accounts
252			f = open(os.path.join(self.configDir,"accounts.txt"), "w")
253			for account in self.accounts.keys():
254			account = self.accounts[account]
255			print >>f, "%s\t%s\t%s\t%s" % (
256			account.nick.encode("utf-8"),
257			account.login.encode("utf-8"),
258			account.passwd.encode("utf-8"),
259			account.email.encode("utf-8")
260			)
261			f.close()
262			return None, None
263
264			def serverShutdown(self, sid):
265			# check admin
266			session = self.getSession(sid)
267			if session.login != ADMIN_LOGIN:
268			raise SecurityException('You cannot issue this command.')
269			log.message('Shutting down server')
270			import ige.RPCServer
271			ige.RPCServer.running = 0
272			return 1, None
273
274			# new rpc interface wrappers
275			def rpc_hello(self, login, clientId):
276			return self.hello(None, login, clientId)[0]
277
278			def rpc_login(self, sid, login, cpasswd, hostID):
279			return self.login(sid, login, cpasswd, hostID)[0]
280
281			def rpc_logout(self, sid):
282			return self.logout(sid)[0]
283
284			def rpc_createAccount(self, sid, login, passwd, nick, email):
285			return self.createAccount(sid, login, passwd, nick, email)[0]
286
287			# Session class keeps various session data
288			class Session:
289
290			def __init__(self, sid):
291			self.cid = None
292			self.sid = sid
293			self.messages = {}
294			self.touch()
295
296			def setAttrs(self, login, nick, email):
297			self.login = login
298			self.nick = nick
299			self.email = email
300
301			def touch(self):
302			# 10 minutes timeout of session
303			self.timeout = time.time() + 10 * 60
304
305			class Mapping(dict):
306			pass
307

ospaceteam / outerspace

Pull Request — master (#192)

ige.ClientMngr.ClientMngr.resetAIAccounts() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like