# along with this program. If not, see <http://www.gnu.org/licenses/>.
import datetime
+import string
from oauthlib.oauth1 import (RequestTokenEndpoint, AuthorizationEndpoint,
AccessTokenEndpoint)
-
+
from mediagoblin.decorators import require_active_login
from mediagoblin.tools.translate import pass_to_ugettext
from mediagoblin.meddleware.csrf import csrf_exempt
from mediagoblin.tools.request import decode_request
-from mediagoblin.tools.response import (render_to_response, redirect,
+from mediagoblin.tools.response import (render_to_response, redirect,
json_response, render_400,
form_response)
from mediagoblin.tools.crypto import random_string
from mediagoblin.db.models import NonceTimestamp, Client, RequestToken
# possible client types
-client_types = ["web", "native"] # currently what pump supports
+CLIENT_TYPES = ["web", "native"] # currently what pump supports
+OAUTH_ALPHABET = (string.ascii_letters.decode('ascii') +
+ string.digits.decode('ascii'))
@csrf_exempt
def client_register(request):
""" Endpoint for client registration """
try:
- data = decode_request(request)
+ data = decode_request(request)
except ValueError:
error = "Could not decode data."
return json_response({"error": error}, status=400)
if "type" not in data:
error = "No registration type provided."
return json_response({"error": error}, status=400)
- if data.get("application_type", None) not in client_types:
+ if data.get("application_type", None) not in CLIENT_TYPES:
error = "Unknown application_type."
return json_response({"error": error}, status=400)
-
+
client_type = data["type"]
if client_type == "client_update":
return json_response({"error": error}, status=400)
client = Client.query.filter_by(
- id=data["client_id"],
+ id=data["client_id"],
secret=data["client_secret"]
).first()
return json_response({"error": error}, status=403)
client.application_name = data.get(
- "application_name",
+ "application_name",
client.application_name
)
)
app_name = ("application_type", client.application_name)
- if app_name in client_types:
+ if app_name in CLIENT_TYPES:
client.application_name = app_name
elif client_type == "client_associate":
return json_response({"error": error}, status=400)
# generate the client_id and client_secret
- client_id = random_string(22) # seems to be what pump uses
- client_secret = random_string(43) # again, seems to be what pump uses
+ client_id = random_string(22, OAUTH_ALPHABET)
+ client_secret = random_string(43, OAUTH_ALPHABET)
expirey = 0 # for now, lets not have it expire
expirey_db = None if expirey == 0 else expirey
- application_type = data["application_type"]
-
+ application_type = data["application_type"]
+
# save it
client = Client(
- id=client_id,
- secret=client_secret,
+ id=client_id,
+ secret=client_secret,
expirey=expirey_db,
application_type=application_type,
)
if logo_url is not None and not validate_url(logo_url):
error = "Logo URL {0} is not a valid URL.".format(logo_url)
return json_response(
- {"error": error},
+ {"error": error},
status=400
)
else:
client.logo_url = logo_url
-
+
client.application_name = data.get("application_name", None)
contacts = data.get("contacts", None)
# not a valid email
error = "Email {0} is not a valid email.".format(contact)
return json_response({"error": error}, status=400)
-
-
+
+
client.contacts = contacts
redirect_uris = data.get("redirect_uris", None)
client.redirect_uri = redirect_uris
-
+
client.save()
expirey = 0 if client.expirey is None else client.expirey
def request_token(request):
""" Returns request token """
try:
- data = decode_request(request)
+ data = decode_request(request)
except ValueError:
error = "Could not decode data."
return json_response({"error": error}, status=400)
if not data and request.headers:
data = request.headers
-
+
data = dict(data) # mutableifying
authorization = decode_authorization_header(data)
return form_response(tokens)
-@require_active_login
+@require_active_login
def authorize(request):
""" Displays a page for user to authorize """
if request.method == "POST":
return authorize_finish(request)
-
+
_ = pass_to_ugettext
token = request.args.get("oauth_token", None)
if token is None:
if oauth_request is None:
err_msg = _("No request token found.")
return render_400(request, err_msg)
-
+
if oauth_request.used:
return authorize_finish(request)
-
+
if oauth_request.verifier is None:
orequest = GMGRequest(request)
request_validator = GMGRequestValidator()
"mediagoblin/api/authorize.html",
context
)
-
+
def authorize_finish(request):
""" Finishes the authorize """
verifier = request.form["oauth_verifier"]
oauth_request = RequestToken.query.filter_by(token=token, verifier=verifier)
oauth_request = oauth_request.first()
-
+
if oauth_request is None:
# invalid token or verifier
err_msg = _("No request token found.")
@csrf_exempt
def access_token(request):
- """ Provides an access token based on a valid verifier and request token """
+ """ Provides an access token based on a valid verifier and request token """
data = request.headers
- parsed_tokens = decode_authorization_header(data)
+ parsed_tokens = decode_authorization_header(data)
if parsed_tokens == dict() or "oauth_token" not in parsed_tokens:
error = "Missing required parameter."
projects / mediagoblin.git / blobdiff