Coverage for integrations / social / privacy.py: 0.0%
42 statements
« prev ^ index » next coverage.py v7.14.0, created at 2026-05-12 04:49 +0000
« prev ^ index » next coverage.py v7.14.0, created at 2026-05-12 04:49 +0000
1"""
2HevolveSocial — Per-post privacy gate.
4Phase 7c.5. Plan reference: sunny-gliding-eich.md, Part E.10.
6Privacy levels:
7 - 'public' visible to everyone (default; existing rows = NULL = public)
8 - 'friends' visible only to author + active-friends of author
9 - 'community' visible only to author + members of post.community_id
10 - 'private' visible only to author
12NULL is normalised to 'public' for backward compatibility — every
13post that existed before v48 stays exactly as visible as it was.
15Two surfaces:
16 - can_view_post(db, viewer_user, post)
17 Single-post authoritative check. Used by GET /posts/<id>.
18 - visible_posts_filter(viewer_user)
19 SQLAlchemy filter expression for use in list queries
20 (PostService.list_posts). Pre-filters at SQL for performance.
22Hidden / soft-deleted posts are still gated by is_hidden / is_deleted
23in the caller. privacy.py does NOT override moderator hide or
24soft delete; it only enforces the user-set privacy level.
26Transport: N/A — read-only synchronous helpers used by api.py
27endpoints. No fan-out implications: privacy is applied at READ
28time, not at write time, so existing MessageBus.publish call sites
29need no changes.
31Block interaction: separate concern. The Block table tears down
32the PeerLink trust ratchet (Plan R.5) and is enforced by
33FriendService.is_friend (which returns False for blocked pairs).
34A blocked viewer still falls through public_clause for public
35posts; explicit feed-level Block filtering is a 7c.6 follow-up
36tracked in Plan W (the existing services.py FollowService and
37encounter visibility paths already do this for their surfaces).
38"""
40import logging
42from sqlalchemy import or_, text
44logger = logging.getLogger('hevolve_social')
47PRIVACY_LEVELS = ('public', 'friends', 'community', 'private')
50def _normalize(level):
51 """Bad input → 'public'. Used by both surfaces below so the
52 NULL-means-public invariant lives in exactly one place.
53 """
54 if not level:
55 return 'public'
56 if level not in PRIVACY_LEVELS:
57 return 'public'
58 return level
61def can_view_post(db, viewer_user, post) -> bool:
62 """Authoritative single-post check. viewer_user may be None.
64 Returns True iff the viewer is allowed to see this specific post
65 given its privacy level + the viewer's relationship to the author.
67 Order of checks (cheapest first):
68 1. Public → True (everyone, including anonymous).
69 2. Author → True (always sees own posts at any level).
70 3. Private → False (no friend / community escape hatch).
71 4. Friends → FriendService.is_friend(viewer, author).
72 5. Community → membership lookup against post.community_id.
73 """
74 level = _normalize(getattr(post, 'privacy', None))
76 if level == 'public':
77 return True
79 if viewer_user is None:
80 return False
82 if post.author_id == viewer_user.id:
83 return True
85 if level == 'private':
86 return False
88 if level == 'friends':
89 try:
90 from .friend_service import FriendService
91 return FriendService.is_friend(db, viewer_user.id, post.author_id)
92 except Exception as e:
93 logger.warning("privacy.can_view_post friends check failed: %s", e)
94 return False
96 if level == 'community':
97 if not post.community_id:
98 # Misconfigured: community-scope on a post without a
99 # community. Fail safe — treat as private.
100 return False
101 try:
102 row = db.execute(text(
103 "SELECT 1 FROM memberships "
104 "WHERE parent_kind = 'community' AND parent_id = :cid "
105 "AND member_id = :uid LIMIT 1"),
106 {'cid': post.community_id, 'uid': viewer_user.id}
107 ).fetchone()
108 return row is not None
109 except Exception as e:
110 logger.warning(
111 "privacy.can_view_post community check failed: %s", e)
112 return False
114 # Defensive — _normalize should have collapsed unknown levels to
115 # 'public'. Anything reaching here is a programmer bug.
116 return False
119def visible_posts_filter(viewer_user):
120 """Return a SQLAlchemy filter expression to AND into a Post query.
122 Pre-filters at SQL for performance: cuts the candidate set before
123 pagination so we don't fetch 1000 posts to filter to 50. The
124 single-post check (can_view_post) remains the canonical authority
125 and should be re-applied at handler level for edge cases (e.g.,
126 a post that becomes inaccessible between query and serve).
128 Anonymous viewer (None) → public-only.
130 Authenticated viewer → public OR own OR (friends + active friend)
131 OR (community + membership row exists). Friendship + Membership
132 are not ORM models, so the friends/community arms use raw SQL
133 EXISTS subqueries with bound parameters (no string interpolation
134 of user-controlled data — the bind layer escapes everything).
135 """
136 from .models import Post
138 public_clause = or_(Post.privacy.is_(None), Post.privacy == 'public')
139 if viewer_user is None:
140 return public_clause
142 vid = viewer_user.id
144 # 'friends' arm: status='active' on the sorted (a,b) row that
145 # contains both the post author and the viewer. Two arms because
146 # author may be on either side of the sorted pair. The
147 # `author_id <> :viewer` predicate makes the arms mutually
148 # exclusive with own_clause so the EXISTS subquery doesn't fire
149 # for the viewer's own posts (P3-07 reviewer fix).
150 friends_arm = text(
151 "(posts.privacy = 'friends' "
152 " AND posts.author_id <> :__priv_vid_friends "
153 " AND EXISTS ("
154 " SELECT 1 FROM friendships f "
155 " WHERE f.status = 'active' "
156 " AND ((f.user_a_id = posts.author_id "
157 " AND f.user_b_id = :__priv_vid_friends) "
158 " OR (f.user_b_id = posts.author_id "
159 " AND f.user_a_id = :__priv_vid_friends))))"
160 ).bindparams(__priv_vid_friends=vid)
162 # 'community' arm: viewer must have a memberships row for this
163 # post's community. community_id may be NULL (cross-posted /
164 # personal feed posts) — we deliberately do not match those: a
165 # post tagged 'community' must have a community_id, otherwise
166 # it's misconfigured and should not surface.
167 # Same `<> :viewer` exclusion as friends_arm for arm-disjointness.
168 # Memberships rows are hard-deleted on community leave (no
169 # left_at column today — P3-09 contract: if memberships ever
170 # grows a soft-leave column, this EXISTS must be tightened).
171 # Tenant scoping is applied by the existing tenant_filter listener
172 # on the outer Post query; the EXISTS subqueries are NOT
173 # tenant-filtered (memberships and friendships are global tables;
174 # cross-tenant friendship/membership is impossible by other
175 # invariants).
176 community_arm = text(
177 "(posts.privacy = 'community' AND posts.community_id IS NOT NULL "
178 " AND posts.author_id <> :__priv_vid_community "
179 " AND EXISTS ("
180 " SELECT 1 FROM memberships m "
181 " WHERE m.parent_kind = 'community' "
182 " AND m.parent_id = posts.community_id "
183 " AND m.member_id = :__priv_vid_community))"
184 ).bindparams(__priv_vid_community=vid)
186 own_clause = Post.author_id == vid
188 # Note: 'private' is covered by own_clause only (author can always
189 # see their own posts at any level). An explicit private branch
190 # is unnecessary and would just duplicate own_clause.
191 return or_(public_clause, own_clause, friends_arm, community_arm)
194__all__ = ['PRIVACY_LEVELS', 'can_view_post', 'visible_posts_filter']