from __future__ import annotations from datetime import datetime from typing import List from grung.types import BackReference, Collection, DateTime, Field, Password, Pointer, Record, Timestamp from tinydb import Query class Page(Record): """ A page in the wiki. Just about everything in the databse is either a Page or a subclass of a Page. """ @classmethod def fields(cls): return [ *super().fields(), # Pick up the UID and whatever other non-optional fields exist Field("uri", unique=True), # The URI for the page, relative to the app's VIEW_URI Field("name"), # The portion of the URI after the last / Field("title"), # The page title Field("body"), # The main content blob of the page Collection("members", Page), # The pages that exist below this page's URI BackReference("parent", value_type=Page), # The page that exists above this page's URI Pointer("author", value_type=User), # The last user to touch the page. DateTime("created"), # When the page was created Timestamp("last_modified"), # The last time the page was modified. Collection("acl", Permissions), ] def before_insert(self, db): """ Make the following adjustments before saving this record: * Derive the name from the title, or the title from the name * Derive the URI from the hierarchy of the parent. """ super().before_insert(db) now = datetime.utcnow() if not self.doc_id and self.created < now: self.created = now if not self.name and not self.title: raise Exception("Must provide either a name or a title!") if not self.name: self.name = self.title.title().replace(" ", "") if not self.title: self.title = self.name self.uri = (self.parent.uri + "/" if self.parent and self.parent.uri != "/" else "") + self.name def after_insert(self, db): """ After saving this record, ensure that any page in the members collection is updated with the correct URI. This ensures that if a page is moved from one collection to another, the URI is updated. """ super().after_insert(db) if not hasattr(self, "members"): return for child in self.members: obj = BackReference.dereference(child, db) obj.uri = f"{self.uri}/{obj.name}" child = db.save(obj) def get_child(self, obj: Record): for page in self.members: if page.uid == obj.uid: return page return None def set_permissions(self, entity: Entity, permissions: List, db) -> Record: perms = db.save(Permissions(entity=entity, grants="".join(permissions))) self.acl = [entry for entry in self.acl if entry.entity != entity] + [perms] self = db.save(self) return perms class Entity(Page): def has_permission(self, record: Record, requested: str, db) -> bool: # if there's no ACL at all, the record is world-readable. if not getattr(record, "acl", None): return requested == Permissions.READ # Use the grant specific to this entity, if there is one for entry in record.acl: if entry.entity.uid == self.uid: return requested in entry.grants for group in db.Group.search(Query()["members"].any([self.reference])): if group.has_permission(record, requested, db): return True return False def can_read(self, record: Record, db): return self.has_permission(record, Permissions.READ, db) def can_write(self, record: Record, db): return self.has_permission(record, Permissions.WRITE, db) def can_delete(self, record: Record, db): return self.has_permission(record, Permissions.DELETE, db) class User(Entity): """ A website user, editable as a wiki page. """ def check_credentials(self, username: str, password: str) -> bool: return username == self.name and self._metadata.fields["password"].compare(password, self.password) @classmethod def fields(cls): return [ field for field in [ *super().fields(), Field("email", unique=True), Password("password"), ] if field.name != "members" ] class Group(Entity): """ A set of users, editable as a wiki page. """ class NPC(Page): """ An NPC, editable as a wiki page. """ class Permissions(Record): READ = "r" WRITE = "w" DELETE = "d" @classmethod def fields(cls): return [*super().fields(), Pointer("entity", Entity), Field("grants")]