Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
user = addr.split('@')[0]
source = FacebookEmailAccount.query(FacebookEmailAccount.email_user == user).get()
logging.info('Source for %s is %s', user, source)
htmls = list(body.decode() for _, body in email.bodies('text/html'))
fbe = FacebookEmail.get_or_insert(
message_id, source=source.key if source else None, htmls=htmls)
logging.info('FacebookEmail created %s: %s', fbe.created, fbe.key.urlsafe())
if not source:
self.response.status_code = 404
self.response.write('No Facebook email user found with address %s' % addr)
return
for html in htmls:
obj = gr_facebook.Facebook().email_to_object(html)
if obj:
break
else:
self.response.status_code = 400
self.response.write('No HTML body could be parsed')
return
logging.info('Converted to AS1: %s', json_dumps(obj, indent=2))
base_obj = source.gr_source.base_object(obj)
# note that this ignores the id query param (the post's user id) and uses
# the source object's user id instead.
base_obj['url'] = source.canonicalize_url(base_obj['url'])
# also note that base_obj['id'] is not a tag URI, it's the raw Facebook post
# id, eg '104790764108207'. we don't use it from activities_json much,
# though, just in PropagateResponse.source_url(), which handles this fine.
def get_comment(self, id, **kwargs):
resp = ndb.Key('Response', self.gr_source.tag_uri(id))
email = FacebookEmail.query(FacebookEmail.response == resp).get()
if email:
return gr_facebook.Facebook().email_to_object(email.htmls[0])
MAX_RESOLVED_OBJECT_IDS = 200
MAX_POST_PUBLICS = 200
# empirically we've seen global user ids as high as 407874323168, and app scoped
# ids as low as 527127880724, so there's probably not a single cutoff like this.
# but it's ok as an approximation.
MIN_APP_SCOPED_ID = 500000000000
class FacebookPage(models.Source):
"""A Facebook profile or page.
The key name is the Facebook id.
"""
GR_CLASS = gr_facebook.Facebook
OAUTH_START_HANDLER = oauth_facebook.StartHandler
SHORT_NAME = 'facebook'
URL_CANONICALIZER = util.UrlCanonicalizer(
domain=GR_CLASS.DOMAIN,
subdomain='www',
query=True,
approve=r'https://www\.facebook\.com/[^/?]+/posts/[^/?]+$',
headers=util.REQUEST_HEADERS)
# no reject regexp; non-private FB post URLs just 404
# unique name used in FB URLs, e.g. facebook.com/[username]
username = ndb.StringProperty()
# inferred from syndication URLs if username isn't available
inferred_username = ndb.StringProperty()
# inferred application-specific user IDs (from other applications)
"""Returns the 'base' silo object that an object operates on.
This is mostly a big bag of heuristics for reverse engineering and
parsing Facebook URLs. Whee.
Args:
obj: ActivityStreams object
verb: string, optional
resolve_numeric_id: if True, tries harder to populate the numeric_id field
by making an additional API call to look up the object if necessary.
Returns:
dict, minimal ActivityStreams object. Usually has at least id,
numeric_id, and url fields; may also have author.
"""
base_obj = super(Facebook, self).base_object(obj)
url = base_obj.get('url')
if not url:
return base_obj
author = base_obj.setdefault('author', {})
base_id = base_obj.get('id')
if base_id and not base_obj.get('numeric_id'):
if util.is_int(base_id):
base_obj['numeric_id'] = base_id
elif resolve_numeric_id:
base_obj = self.user_to_actor(self.urlopen(base_id))
try:
parsed = urllib.parse.urlparse(url)
params = urllib.parse.parse_qs(parsed.query)
def base_object(self, obj, verb=None, resolve_numeric_id=False):
"""Returns the 'base' silo object that an object operates on.
This is mostly a big bag of heuristics for reverse engineering and
parsing Facebook URLs. Whee.
Args:
obj: ActivityStreams object
verb: string, optional
resolve_numeric_id: if True, tries harder to populate the numeric_id field
by making an additional API call to look up the object if necessary.
Returns: dict, minimal ActivityStreams object. Usually has at least id,
numeric_id, and url fields; may also have author.
"""
base_obj = super(Facebook, self).base_object(obj)
url = base_obj.get('url')
if not url:
return base_obj
author = base_obj.setdefault('author', {})
base_id = base_obj.get('id')
if base_id and not base_obj.get('numeric_id'):
if util.is_int(base_id):
base_obj['numeric_id'] = base_id
elif resolve_numeric_id:
base_obj = self.user_to_actor(self.urlopen(base_id))
try:
parsed = urlparse.urlparse(url)
params = urlparse.parse_qs(parsed.query)
def parsed_post_id(id):
parsed = gr_facebook.Facebook.parse_id(id)
return parsed.post if parsed.post else id
The key id is the Message-ID header.
"""
source = ndb.KeyProperty()
htmls = ndb.TextProperty(repeated=True)
created = ndb.DateTimeProperty(auto_now_add=True, required=True)
response = ndb.KeyProperty()
class FacebookEmailAccount(FacebookPage):
"""A Facebook profile or page.
The key name is the Facebook user id.
"""
GR_CLASS = gr_facebook.Facebook
SHORT_NAME = 'facebook-email'
# username for the inbound email address that users forward notification
# emails to. the address will be [email_user]@brid-gy.appspotmail.com.
# https://cloud.google.com/appengine/docs/standard/python/mail/receiving-mail-with-mail-api
email_user = ndb.StringProperty()
def get_activities_response(self, **kwargs):
activities = []
activity_id = kwargs.get('activity_id')
if activity_id:
activities = [{
'id': activity_id,
'url': 'https://www.facebook.com/%s/posts/%s' % (self.key.id(), activity_id),
}]
def new(handler, auth_entity=None, **kwargs):
"""Creates and returns a :class:`FacebookPage` for the logged in user.
Args:
handler: the current :class:`webapp2.RequestHandler`
auth_entity: :class:`oauth_dropins.facebook.FacebookAuth`
kwargs: property values
"""
user = json_loads(auth_entity.user_json)
gr_source = gr_facebook.Facebook(access_token=auth_entity.access_token())
actor = gr_source.user_to_actor(user)
return FacebookPage(id=user['id'],
auth_entity=auth_entity.key,
name=actor.get('displayName'),
username=actor.get('username'),
picture=actor.get('image', {}).get('url'),
url=actor.get('url'),
**kwargs)
'jsonfeed',
'mf2-json',
)
SILOS = (
'flickr',
'github',
'instagram',
'mastodon',
'twitter',
)
OAUTHS = { # maps oauth-dropins module name to module
name: importlib.import_module('oauth_dropins.%s' % name)
for name in SILOS
}
SILO_DOMAINS = {cls.DOMAIN for cls in (
Facebook,
Flickr,
GitHub,
Instagram,
Twitter,
)}
SCOPE_OVERRIDES = {
# https://developers.facebook.com/docs/reference/login/
'facebook': 'user_status,user_posts,user_photos,user_events',
# https://developer.github.com/apps/building-oauth-apps/scopes-for-oauth-apps/
'github': 'notifications,public_repo',
# https://docs.joinmastodon.org/api/permissions/
'mastodon': 'read',
}
class FrontPageHandler(handlers.TemplateHandler):