ru3dWq78

1#!/usr/local/bin/python
2from kyotocabinet import *
3import sys
4from db_response_pb2 import *
5from db_request_pb2 import *
6from objects_pb2 import *
7import bcrypt
8import ProtocolBufferRPCServer
9
10# TODO: Move all the kyoto cabinet-specific functionality into
11# its own class.
12
13def error(err_str):
14    print >>sys.stderr, err_str
15
16# Some constants
17db_exchange = "db_xchg"
18db_queue = "db_queue"
19dbname = "muckcasket.kch"
20playerdbname = "muckplayercasket.kch"
21SITE_ID = 0 # No sites implemented yet.
22NEXT_ID_KEY = "next_id"
23
24def printable_dbref(dbref):
25    """This is a utility function to convert a dbref into a printable string.
26    Presently unused until I see if Message.__str__() is unsuitable.
27    """
28    return "%u:%u" % dbref.site_id, dbref.object_id
29
30def confirm_player_admin(response, dbref):
31    """Checks that dbref is in the database, that is a player,
32    and that the player has the admin tag. If any check fails,
33    response is filled out with an appropriate error message.
34    """
35    global db
36    player_bytes = db.get(dbref)
37    if not playerBytes:
38        setup_status_response(response,
39            object_id=request.player_dbref,
40            message="Couldn't find dbref %s in database" % 
41            str(request.player_dbref))
42        return False
43    player = ObjectBase()
44    player.ParseFromString(player_bytes)
45    if (not player.type == PLAYER or 
46        not player.player.administrator):
47        setup_status_response(response,
48            object_id=request.player_dbref,
49            message="Invalid player dbref in delete command: %s " %
50            str(request.player_dbref))
51        return False
52    return True
53
54def setup_status_response(response, success=False, object_id=None,
55                          message=""):
56    """Because we use Status responses for errors in many places,"""
57    """we use this utility function to set up their fields."""
58    response.type = STATUS
59    response.status_fields.success = success
60    if object_id:
61        response.status_fields.object_id.CopyFrom(object_id)
62    response.status_fields.message = message
63
64def handle_dbrefs(request, response):
65    """ Handle a request for all objects of a particular type."""
66    global db
67    
68    response.type = LIST_OF_DBREFS
69    target_type = request.request_dbrefs_fields.type_to_get
70    
71    # Do a full scan of the database for this implementation.
72    # This is nothing short of /terrible/ as it reads the entire database.
73    obj = ObjectBase()
74    for key in db:
75        obj.ParseFromString(db[key])
76        if obj.type == target_type:
77            response.list_of_dbrefs_fields.object_ids.append(obj.dbref)
78
79def handle_objects(request, response):
80    global db
81    response.type = OBJECTS_RESP
82    for obj_id in request.request_objects_fields.object_ids:
83        obj_bytes = db.get(obj_id.SerializeToString())
84        if obj_bytes:
85            obj = ObjectBase()
86            obj.ParseFromString(obj_bytes)
87            response.object_fields.objects.append(obj)
88        else:
89            failure = ResponseStatus()
90            failure.success=False
91            failure.object_id = obj_id
92            response.object_fields.failed_objects.append(failure)
93
94def handle_player(request, response):
95    global db, playerdb
96    # Attempt to look up the player in the playerdb.
97    # This is just the object_id.
98    dbref = playerdb.get(request.requestPlayerObjectFields.player_name)
99    if dbref:
100        objBytes = db.get(dbref)
101        if not objBytes:
102            setUpStatusResponse(response, 
103                object_id=Dbref().ParseFromString(dbref),
104                message="Couldn't find db entry for player %s" % 
105                request.request_player_object_fields.player_name)
106        else:
107            obj = ObjectBase().ParseFromString(objBytes)
108            if not obj.type == PLAYER:
109                setup_status_response(response,   
110                    object_id=Dbref().ParseFromString(dbref),
111                    message="Db entry for player %s not a player object" % 
112                    request.request_player_object_fields.player_name)
113            else:                                  
114                response.type = PLAYER_RESP
115                response.player_fields.player = obj
116
117def handle_delete(request, response):
118    """Delete an entry in the database. I strongly suspect we have subtle race
119    conditions herein. Also we want to add more formal permissions checking.
120    """
121    global db, playerdb
122    obj_bytes = db.get(
123        request.delete_object_fields.object_id.SerializeToString())
124    if not obj_bytes:
125        setup_status_response(response,
126            object_id=request.delete_object_fields.object_id,
127            message="Couldn't find dbref %s in database" % 
128            str(request.delete_object_fields.object_id))
129        return
130
131    obj = ObjectBase()
132    obj.ParseFromString(obj_bytes)
133    if obj.type == PLAYER:
134        # If the object is a player check the player field for the player
135        # requesting a delete.
136        # Only admins can delete other players (including themselves....)
137        if not confirm_player_admin(response, request.player_dbref):
138            return
139        if not db.remove(request.delete_object_fields.object_id):
140            setup_status_response(response,
141                object_id=request.delete_object_fields.object_id,
142                message="Unknown error while deleting %s" %
143                str(request.delete_object_fields.object_id))
144        else:
145            setup_status_response(response, success=True,
146                object_id=request.delete_object_fields.object_id)
147
148def handle_create_player(request, response):
149    """Adds a new player to the system. This should eventually check 
150    for admin status on the requesting player but for now, we want to
151    get things in the database easily.
152    There's also a race condition herein wherein the same player can be 
153    created by two threads at once. The consequences are not drastic,
154    but only because all we set as of yet is the password.
155    We're also assuming the player name comes in 'clean'.
156    Also if we die halfway, we might reserve the name without finishing
157    setting up its entry.
158    """
159    global db, playerdb
160    if playerdb.get(request.create_player_fields.name):
161        setup_status_response(response,
162            message="Player %s already exists" %
163            request.create_player_fields.name)
164        return
165    
166    # Make a dbref for the player and an object.
167    object_id = db.increment(NEXT_ID_KEY, 1, 0)
168
169    player_object = ObjectBase()
170    player_object.dbref.site_id = SITE_ID
171    player_object.dbref.object_id = object_id
172    player_object.name = request.create_player_fields.name
173    player_object.description = ""
174    player_object.type = PLAYER
175    player_object.player.password = bcrypt.hashpw(
176        request.create_player_fields.password, bcrypt.gensalt())
177    player_object.player.administrator = (
178        request.create_player_fields.administrator)
179    
180    if not db.add(player_object.dbref.SerializeToString(),
181                  player_object.SerializeToString()):
182        setup_status_response(response, message = 
183            "Couldn't add player %s to database" % player_object.name)
184        return
185    if not playerdb.add(player_object.name,
186                        player_object.dbref.SerializeToString()):
187        setup_status_response(response, message =
188            "Couldn't add player %s to player table. May have"
189            " been a collission" % player_object.name)
190        # TODO: If we believe this situation to be common, also
191        # delete the player_object from db. For now we'll keep
192        # it for debugging.
193        return
194    setup_status_response(response, success=True, object_id=player_object.dbref)
195
196def handle_set_object_data(request, response):
197    """Take out the existing object, update its fields with the new
198    data and stick it back in. If the record inside the database has
199    changed while we performed the update, back out.
200    TODO: Use revision numbers? This is if we care about this situation:
201    User1 wants to mutate a->a'.
202    Between the time the user1 retrieves a and writes a', other users
203    perform a->b->a.
204    User1 sees that a is still in the DB, makes the modification unaware
205    that the object was mutated twice in between.
206    Zeta: I don't think this is that big a deal.
207    TODO: Better testing for object type.
208    TODO: We might want to consider changing SetObjectData to be a partial
209    ObjectBase
210    """
211    global db
212    
213    # Set up a mutation object
214    obj_mut = ObjectBase()
215    if request.set_object_data_fields.HasField("object_name"):
216        obj_mut.name = request.set_object_data_fields.object_name
217    if request.set_object_data_fields.HasField("object_description"):
218        obj_mut.description = (
219            request.set_object_data_fields.object_description)
220    if request.set_object_data_fields.HasField("password"):
221        obj_mut.player.password = bcrypt.hashpw(
222            request.set_object_data_fields.password, bcrypt.gensalt())
223    if request.set_object_data_fields.HasField("administrator"):
224        obj_mut.player.administrator = (
225            request.set_object_data_fields.administrator)
226    
227    # Get the existing object and apply the changes.
228    key = request.set_object_data_fields.object_id.SerializeToString()
229    obj_bytes = db.get(key)
230    if not obj_bytes:
231        setup_status_response(response,
232            object_id=request.set_object_data_fields.object_id,
233            message="Couldn't find object for %s in database" %
234            str(request.set_object_data_fields.object_id))
235        return
236    obj = ObjectBase()
237    obj.ParseFromString(obj_bytes)
238    obj.MergeFrom(obj_mut)
239    
240    # Attempt to save, error out if the object has changed.
241    if not db.cas(key, obj_bytes, obj.SerializeToString()):
242        setup_status_response(response,
243            object_id=request.set_object_data_fields.object_id,
244            message="Object %s changed while we were applying changes" %
245            str(request.set_object_data_fields.object_id))
246    else:
247        setup_status_response(response, success=True,
248            object_id=request.set_object_data_fields.object_id)
249
250# A hash of callbacks to handle various message types
251message_handlers = { DBREFS : handle_dbrefs,
252                     PLAYER_REQ : handle_player,
253                     OBJECTS_REQ : handle_objects,
254                     DELETE : handle_delete,
255                     CREATE_PLAYER : handle_create_player,
256                     CREATE_PROGRAM : None,
257                     SET_OBJECT_DATA : handle_set_object_data,
258                   }
259
260def process_message(pbrpcserver, message, token):
261    response = DBResponse()
262    message_handlers[message.type](message, response)
263    pbrpcserver.send_reply(response, token)
264
265
266# Open our databases
267db = DB()
268playerdb = DB()
269if not db.open(dbname, DB.OWRITER | DB.OCREATE):
270    error("Couldn't open db " + dbname + " " + str(db.error()))
271    exit
272if not playerdb.open(playerdbname, DB.OWRITER | DB.OCREATE):
273    error("Couldn't open db " + playerdbname + " " + str(playerdb.error()))
274
275# Create a global server for our Protocol Buffer stuff
276pbrpcserver = ProtocolBufferRPCServer.ProtocolBufferRPCServer(
277    request_class=RequestMessage,
278    exchange=db_exchange,
279    routing_key=db_queue,
280    message_handler=process_message)
281
282pbrpcserver.start()