discours.io/core
5
0
Fork 0
Code Pull Requests Actions Projects Activity

forbidden-fix
Some checks failed
Deploy on push / deploy (push) Failing after 20s
Details

Browse Source
This commit is contained in:
Untone 2024-03-06 10:44:08 +03:00
parent b09ea39668
commit cb535cffea
1 changed files with 44 additions and 52 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

96
resolvers/editor.py
View File

@ -33,13 +33,12 @@ async def get_my_shout(_, info, shout_id: int):
user_id = info.context.get('user_id', '') user_id = info.context.get('user_id', '')
roles = info.context.get('roles', []) roles = info.context.get('roles', [])
if not user_id: if not user_id:
error = 'user is not logged in' error = 'unauthorized'
elif shout.created_by != author.id: else:
error = 'author cannot edit this post' if 'editor' in roles or filter(lambda x: x.id == author.id, [x for x in shout.authors]):
elif 'editor' not in roles: return {"error": error, "shout": shout}
error = 'user has no editor role' else:
elif not any([x.id == author.id for x in shout.authors]): error = 'forbidden'
error = 'author have no permissions to read this not published shout'
return {"error": error, "shout": shout} return {"error": error, "shout": shout}
@ -232,40 +231,37 @@ async def update_shout(_, info, shout_id: int, shout_input=None, publish=False):
) )
shout_input['slug'] = slug shout_input['slug'] = slug
if ( if filter(lambda x: x.id == author.id, [x for x in shout_by_id.authors]) or 'editor' in roles:
shout_by_id.created_by != author.id # topics patch
and not filter(lambda x: x == author.id, shout_by_id.authors) topics_input = shout_input.get('topics')
and 'editor' not in roles if topics_input:
): patch_topics(session, shout_by_id, topics_input)
return {'error': 'access denied'} del shout_input['topics']
# topics patch # main topic
topics_input = shout_input.get('topics') main_topic = shout_input.get('main_topic')
if topics_input: if main_topic:
patch_topics(session, shout_by_id, topics_input) patch_main_topic(session, main_topic, shout_by_id)
del shout_input['topics']
# main topic shout_input['updated_at'] = current_time
main_topic = shout_input.get('main_topic') shout_input['published_at'] = current_time if publish else None
if main_topic: Shout.update(shout_by_id, shout_input)
patch_main_topic(session, main_topic, shout_by_id) session.add(shout_by_id)
session.commit()
shout_input['updated_at'] = current_time shout_dict = shout_by_id.dict()
shout_input['published_at'] = current_time if publish else None
Shout.update(shout_by_id, shout_input)
session.add(shout_by_id)
session.commit()
shout_dict = shout_by_id.dict() if not publish:
await notify_shout(shout_dict, 'update')
else:
await notify_shout(shout_dict, 'published')
# search service indexing
search_service.index(shout_by_id)
if not publish: return {'shout': shout_dict, 'error': None}
await notify_shout(shout_dict, 'update')
else: else:
await notify_shout(shout_dict, 'published') return {'error': 'access denied', 'shout': None}
# search service indexing
search_service.index(shout_by_id)
return {'shout': shout_dict}
except Exception as exc: except Exception as exc:
logger.error(exc) logger.error(exc)
logger.error(f' cannot update with data: {shout_input}') logger.error(f' cannot update with data: {shout_input}')
@ -275,7 +271,7 @@ async def update_shout(_, info, shout_id: int, shout_input=None, publish=False):
@mutation.field('delete_shout') @mutation.field('delete_shout')
@login_required @login_required
async def delete_shout(_, info, shout_id): async def delete_shout(_, info, shout_id: int):
user_id = info.context.get('user_id') user_id = info.context.get('user_id')
roles = info.context.get('roles') roles = info.context.get('roles')
if user_id: if user_id:
@ -285,25 +281,21 @@ async def delete_shout(_, info, shout_id):
if not shout: if not shout:
return {'error': 'invalid shout id'} return {'error': 'invalid shout id'}
if author and shout: if author and shout:
if ( # NOTE: only owner and editor can mark the shout as deleted
shout.created_by is not author.id if shout.created_by == author.id or 'editor' in roles:
and author.id not in shout.authors for author_id in shout.authors:
and 'editor' not in roles reactions_unfollow(author_id, shout_id)
):
shout_dict = shout.dict()
shout_dict['deleted_at'] = int(time.time())
Shout.update(shout, shout_dict)
session.add(shout)
session.commit()
await notify_shout(shout_dict, 'delete')
return {'error': None}
else:
return {'error': 'access denied'} return {'error': 'access denied'}
for author_id in shout.authors:
reactions_unfollow(author_id, shout_id)
shout_dict = shout.dict()
shout_dict['deleted_at'] = int(time.time())
Shout.update(shout, shout_dict)
session.add(shout)
session.commit()
await notify_shout(shout_dict, 'delete')
return {}
def handle_proposing(session, r, shout): def handle_proposing(session, r, shout):
if is_positive(r.kind): if is_positive(r.kind):