Add OSM login

This commit is contained in:
Edward Betts 2021-06-16 15:42:04 +02:00
parent b18c846af9
commit 735e833fc9
3 changed files with 210 additions and 4 deletions

View file

@ -2,14 +2,15 @@ from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.schema import ForeignKey, Column from sqlalchemy.schema import ForeignKey, Column
from sqlalchemy.orm import relationship, column_property, deferred from sqlalchemy.orm import relationship, column_property, deferred
from sqlalchemy import func from sqlalchemy import func
from sqlalchemy.types import Integer, String, Float from sqlalchemy.types import Integer, String, Float, Boolean, DateTime, Text
from sqlalchemy.dialects import postgresql from sqlalchemy.dialects import postgresql
from sqlalchemy.sql.expression import cast from sqlalchemy.sql.expression import cast
from sqlalchemy.ext.hybrid import hybrid_property from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy.ext.declarative import declared_attr from sqlalchemy.ext.declarative import declared_attr
from geoalchemy2 import Geography, Geometry from geoalchemy2 import Geography, Geometry
from collections import defaultdict from collections import defaultdict
from .database import session from flask_login import UserMixin
from .database import session, now_utc
from . import wikidata, utils from . import wikidata, utils
import json import json
import re import re
@ -263,3 +264,30 @@ class Polygon(MapMixin, Base):
@hybrid_property @hybrid_property
def area_in_sq_km(self): def area_in_sq_km(self):
return self.area / (1000 * 1000) return self.area / (1000 * 1000)
class User(Base, UserMixin):
__tablename__ = 'user'
id = Column(Integer, primary_key=True)
username = Column(String)
password = Column(String)
name = Column(String)
email = Column(String)
active = Column(Boolean, default=True)
sign_up = Column(DateTime, default=now_utc())
is_admin = Column(Boolean, default=False)
description = Column(Text)
img = Column(String) # OSM avatar
languages = Column(postgresql.ARRAY(String))
single = Column(String)
multi = Column(String)
units = Column(String)
wikipedia_tag = Column(Boolean, default=False)
osm_id = Column(Integer, index=True)
osm_account_created = Column(DateTime)
osm_oauth_token = Column(String)
osm_oauth_token_secret = Column(String)
def is_active(self):
return self.active

77
matcher/osm_oauth.py Normal file
View file

@ -0,0 +1,77 @@
from flask import current_app, session
from requests_oauthlib import OAuth1Session
from urllib.parse import urlencode
from datetime import datetime
from flask import g
from .model import User
from . import user_agent_headers
import lxml.etree
osm_api_base = "https://api.openstreetmap.org/api/0.6"
def api_put_request(path, **kwargs):
user = g.user
assert user.is_authenticated
oauth = OAuth1Session(
current_app.config["CLIENT_KEY"],
client_secret=current_app.config["CLIENT_SECRET"],
resource_owner_key=user.osm_oauth_token,
resource_owner_secret=user.osm_oauth_token_secret,
)
return oauth.request(
"PUT", osm_api_base + path, headers=user_agent_headers(), **kwargs
)
def api_request(path, **params):
user = g.user
assert user.is_authenticated
app = current_app
url = osm_api_base + path
if params:
url += "?" + urlencode(params)
client_key = app.config["CLIENT_KEY"]
client_secret = app.config["CLIENT_SECRET"]
oauth = OAuth1Session(
client_key,
client_secret=client_secret,
resource_owner_key=user.osm_oauth_token,
resource_owner_secret=user.osm_oauth_token_secret,
)
return oauth.get(url, timeout=4)
def parse_iso_date(value):
return datetime.strptime(value, "%Y-%m-%dT%H:%M:%SZ")
def parse_userinfo_call(xml):
root = lxml.etree.fromstring(xml)
user = root[0]
img = user.find(".//img")
account_created = parse_iso_date(user.get("account_created"))
assert user.tag == "user"
return {
"account_created": account_created,
"id": int(user.get("id")),
"username": user.get("display_name"),
"description": user.findtext(".//description"),
"img": (img.get("href") if img is not None else None),
}
def get_username():
if "user_id" not in session:
return # not authorized
user_id = session["user_id"]
user = User.query.get(user_id)
return user.username

View file

@ -1,12 +1,15 @@
#!/usr/bin/python3 #!/usr/bin/python3
from flask import Flask, render_template, request, jsonify, redirect, url_for, g from flask import (Flask, render_template, request, jsonify, redirect, url_for, g,
flash, session)
from sqlalchemy import func, or_ from sqlalchemy import func, or_
from sqlalchemy.orm import selectinload from sqlalchemy.orm import selectinload
from matcher import nominatim, model, database, commons, wikidata, wikidata_api from matcher import nominatim, model, database, commons, wikidata, wikidata_api, osm_oauth
from collections import Counter from collections import Counter
from time import time from time import time
from geoalchemy2 import Geography from geoalchemy2 import Geography
from requests_oauthlib import OAuth1Session
import flask_login
import os import os
import json import json
import GeoIP import GeoIP
@ -19,6 +22,11 @@ app = Flask(__name__)
app.debug = True app.debug = True
app.config.from_object('config.default') app.config.from_object('config.default')
login_manager = flask_login.LoginManager(app)
login_manager.login_view = 'login_route'
osm_api_base = 'https://api.openstreetmap.org/api/0.6'
DB_URL = "postgresql:///matcher" DB_URL = "postgresql:///matcher"
database.init_db(DB_URL) database.init_db(DB_URL)
entity_keys = {"labels", "sitelinks", "aliases", "claims", "descriptions", "lastrevid"} entity_keys = {"labels", "sitelinks", "aliases", "claims", "descriptions", "lastrevid"}
@ -1105,6 +1113,99 @@ def refresh_item(item_id):
return 'done' return 'done'
@app.route('/login')
def login_openstreetmap():
return redirect(url_for('start_oauth',
next=request.args.get('next')))
@app.route('/logout')
def logout():
next_url = request.args.get('next') or url_for('index')
flask_login.logout_user()
flash('you are logged out')
return redirect(next_url)
@app.route('/done/')
def done():
flash('login successful')
return redirect(url_for('index'))
@app.route('/oauth/start')
def start_oauth():
next_page = request.args.get('next')
if next_page:
session['next'] = next_page
client_key = app.config['CLIENT_KEY']
client_secret = app.config['CLIENT_SECRET']
request_token_url = 'https://www.openstreetmap.org/oauth/request_token'
callback = url_for('oauth_callback', _external=True)
oauth = OAuth1Session(client_key,
client_secret=client_secret,
callback_uri=callback)
fetch_response = oauth.fetch_request_token(request_token_url)
session['owner_key'] = fetch_response.get('oauth_token')
session['owner_secret'] = fetch_response.get('oauth_token_secret')
base_authorization_url = 'https://www.openstreetmap.org/oauth/authorize'
authorization_url = oauth.authorization_url(base_authorization_url,
oauth_consumer_key=client_key)
return redirect(authorization_url)
@login_manager.user_loader
def load_user(user_id):
return model.User.query.get(user_id)
@app.route("/oauth/callback", methods=["GET"])
def oauth_callback():
client_key = app.config['CLIENT_KEY']
client_secret = app.config['CLIENT_SECRET']
oauth = OAuth1Session(client_key,
client_secret=client_secret,
resource_owner_key=session['owner_key'],
resource_owner_secret=session['owner_secret'])
oauth_response = oauth.parse_authorization_response(request.url)
verifier = oauth_response.get('oauth_verifier')
access_token_url = 'https://www.openstreetmap.org/oauth/access_token'
oauth = OAuth1Session(client_key,
client_secret=client_secret,
resource_owner_key=session['owner_key'],
resource_owner_secret=session['owner_secret'],
verifier=verifier)
oauth_tokens = oauth.fetch_access_token(access_token_url)
session['owner_key'] = oauth_tokens.get('oauth_token')
session['owner_secret'] = oauth_tokens.get('oauth_token_secret')
r = oauth.get(osm_api_base + '/user/details')
info = osm_oauth.parse_userinfo_call(r.content)
user = model.User.query.filter_by(osm_id=info['id']).one_or_none()
if user:
user.osm_oauth_token = oauth_tokens.get('oauth_token')
user.osm_oauth_token_secret = oauth_tokens.get('oauth_token_secret')
else:
user = model.User(
username=info['username'],
description=info['description'],
img=info['img'],
osm_id=info['id'],
osm_account_created=info['account_created'],
)
database.session.add(user)
database.session.commit()
flask_login.login_user(user)
next_page = session.get('next') or url_for('index_page')
return redirect(next_page)
if __name__ == "__main__": if __name__ == "__main__":
app.run(host="0.0.0.0") app.run(host="0.0.0.0")