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

1""" 

2HevolveSocial — Per-post privacy gate. 

3 

4Phase 7c.5. Plan reference: sunny-gliding-eich.md, Part E.10. 

5 

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 

11 

12NULL is normalised to 'public' for backward compatibility — every 

13post that existed before v48 stays exactly as visible as it was. 

14 

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. 

21 

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. 

25 

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. 

30 

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""" 

39 

40import logging 

41 

42from sqlalchemy import or_, text 

43 

44logger = logging.getLogger('hevolve_social') 

45 

46 

47PRIVACY_LEVELS = ('public', 'friends', 'community', 'private') 

48 

49 

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 

59 

60 

61def can_view_post(db, viewer_user, post) -> bool: 

62 """Authoritative single-post check. viewer_user may be None. 

63 

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. 

66 

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)) 

75 

76 if level == 'public': 

77 return True 

78 

79 if viewer_user is None: 

80 return False 

81 

82 if post.author_id == viewer_user.id: 

83 return True 

84 

85 if level == 'private': 

86 return False 

87 

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 

95 

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 

113 

114 # Defensive — _normalize should have collapsed unknown levels to 

115 # 'public'. Anything reaching here is a programmer bug. 

116 return False 

117 

118 

119def visible_posts_filter(viewer_user): 

120 """Return a SQLAlchemy filter expression to AND into a Post query. 

121 

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). 

127 

128 Anonymous viewer (None) → public-only. 

129 

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 

137 

138 public_clause = or_(Post.privacy.is_(None), Post.privacy == 'public') 

139 if viewer_user is None: 

140 return public_clause 

141 

142 vid = viewer_user.id 

143 

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) 

161 

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) 

185 

186 own_clause = Post.author_id == vid 

187 

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) 

192 

193 

194__all__ = ['PRIVACY_LEVELS', 'can_view_post', 'visible_posts_filter']