Account automation through Python (with Flask) and javascript

Question

This is my first time doing a code review so I expect a ton of criticism on this from all you professionals.

What am I doing?

I'm using python with flask to create a service for my web server. My web server has a login prompt for passing login info into the python service. The request is received as a JSON string element and is converted into a dictionary in python. Then I have my service validate that info to check for any issues with their submission. If any issues were present then I send back a 401 code letting the user know that their login failed. If validation was successful then the user is granted a session cookie and the service creates a thread that stores an object of the thread with the session and Instagram login instance. The user is also sent to the dashboard to control their instance.

What do I need?

I've never been code reviewed before so I really need someone to tear at this. I need to see if there are any issues regarding said code. The code works and there is no runtime issues present.

One of the main concerns is that on line 78 I'm making 2 json.loads executions and all I know is that one doesn't suffice.

Anyways let me know what I should improve.

from flask import *
from flask_wtf.csrf import CSRFProtect
from flask_wtf.csrf import CSRFError
from InstagramAPI import InstagramAPI
from instasession import InstaSession
import threading
import json
import os
import time
import re
app = Flask(__name__)
app.secret_key = os.urandom(32)
csrf = CSRFProtect(app)
app.wtf_csrf_secret_key = os.urandom(32)
csrf.init_app(app)
activeSessions = list() # Active instagram sessions.
#CSRF Protection
@app.errorhandler(CSRFError)
def handle_csrf_error(e):
 return "Invalid CSRF token."
@app.route('/', methods=['GET'])
@app.route('/index', methods=['GET'])
def index():
 if 'user' in session:
 print("Loged In!")
 return redirect(url_for("dashboard"))
 else:
 print("Not Loged In")
 return render_template('index.html')
@app.route('/login', methods=['POST'])
def login():
 print("GOT LOGIN REQUEST")
 
 data = json.loads(request.data)
 username = data["username"]
 password = data["password"]
 
 if validate_username(username) is False:
 abort(401)
 # TODO: Perform check on username to see if instance already exist so we can just resume their instance
 # instead of creating a whole new instance.
 # Perform Instagram Login
 newLogin = InstagramAPI(username, password)
 
 if newLogin.login():
 print("Succesfull Login")
 session['user'] = username # Store session cookie
 # Give the user a thread for their session.
 newThread = threading.Thread(target=process_action, args=(session['user'],))
 newInstance = InstaSession(newThread, session['user'], newLogin)
 newThread.start()
 activeSessions.append( newInstance ) # Store Action Info
 return "success"
 else:
 print("Login Failed")
 abort(401)
#Called whenever the user wants to run a function or get information
@app.route('/action', methods=['POST'])
def action():
 if 'user' in session:
 if request.method == 'POST':
 # Why do we need to use json.loads twice?!
 requestedData = json.loads(json.loads(request.data))
 if "method" in requestedData:
 print(requestedData["method"])
 
 if requestedData["method"] == "autolike":
 print("Auto like Method")
 update_actions(session['user'], requestedData)
 return '{"method" : "auto-like-start"}'
 
 elif requestedData["method"] == "stoplike":
 print("Auto like Method")
 update_actions(session['user'], requestedData)
 return '{"method" : "auto-like-stop"}'
 
 elif requestedData["method"] == "autocomment":
 print("Auto Commenting")
 
 elif requestedData["method"] == "getlikeslog":
 instance = get_current_session( session['user'] )
 if instance is not None:
 return '{"updatelikes" : "'+str(instance.get_logs())+'"}'
 else:
 return "Invalid request!"
 else:
 return "Method does not exist"
 else:
 return "Request is missing method"
 else:
 return redirect(url_for("index"))
@app.route('/dashboard', methods=['GET'])
def dashboard():
 if 'user' in session:
 return render_template('dashboard.html')
 else:
 return redirect(url_for("index"))
@app.route('/logout', methods=['GET'])
def loutout():
 if 'user' in session:
 session.pop('user')
 return "You Loged Out!"
 else:
 return redirect(url_for("index"))
def get_current_session( currentSession ):
 time.sleep(1) #Why? 
 for i in range(len(activeSessions)):
 if activeSessions[i].session == currentSession:
 return activeSessions[i]
 
 print("Could not find a session with users cookie")
 return None
# Updates the variables for the threads.
def update_actions( currentSession, newAction ): # TODO: Use pythonic code format.
 instance = get_current_session( currentSession )
 if instance is not None:
 instance.action = newAction
# Returns the subsession data for this cookie
def get_actions( currentSession ):
 instance = get_current_session( currentSession )
 if instance is not None:
 return instance.action
 
# Threaded function that handles all instagram actions. Variables should be changed on the fly.
def process_action( session ):
 instance = get_current_session(session)
 isDone = False
 action = instance.get_action()
 
 while not isDone:
 isNewAction = False
 time.sleep(1)
 action = instance.get_action()
 
 if action is not None:
 if "method" in action:
 
 if action["method"] == "logout":
 print("Stopping Instance")
 instance.logout()
 
 elif action["method"] == "autolike":
 print("AUTO LIKE")
 print("Tag:"+action["tag"])
 instance.instagram.tagFeed(action["tag"])
 lastJson = instance.instagram.LastJson
 count = len(lastJson["items"])
 for i in range(count):
 if lastJson["items"][i]["has_liked"] is False:
 if isNewAction:
 break
 postID = lastJson["items"][i]["id"]
 instance.instagram.like(postID)
 instance.set_logs(str(postID))
 print("Liked Photo!" + str(postID))
 
 for i in range(45):
 time.sleep(1)
 newAction = instance.get_action()
 
 if action != newAction:
 isNewAction = True
 print("Message is different. Switching to new action")
 break
 
# Performs validation on the username that way we don't get untrusted data.
def validate_username( username ):
 try:
 if len(username) > 30:
 if not re.match(r'^[a-zA-Z0-9._]+$',username):
 if username[0] == '.':
 if username[len(username) - 1] == '.':
 return True
 except ValueError:
 return False

This is the javascript for the dashboard

var csrfToken = null
var logInterval = null
function DisplayMenu( menu ){
 var menus = document.querySelectorAll(".submenu");
 
 for(var i = 0; i < menus.length; i++){
 if(menus[i].id == menu){
 menus[i].style.display = "block";
 } else {
 menus[i].style.display = "none";
 }
 }
 
}
function ProccessResponse(response){
 
 var jsonResponse = JSON.parse(response);
 
 console.log(response.method);
 
 if(jsonResponse.method == "auto-like-start"){
 document.getElementById("like-start").disabled = true;
 document.getElementById("like-stop").disabled = false;
 } else if(jsonResponse.method == "auto-like-stop"){
 document.getElementById("like-start").disabled = false;
 document.getElementById("like-stop").disabled = true;
 clearInterval(logInterval); // Stop logging.
 logInterval = null;
 } else if(jsonResponse.updatelikes) {
 console.log("Got Update!");
 var newEntry = document.getElementById("likes-log").value.concat( "Liked photo:" + jsonResponse.updatelikes + "\n" );
 document.getElementById("likes-log").value = newEntry;
 }
}
function SendRequest( method, endpoint, request ){
 var xhttp = new XMLHttpRequest();
 xhttp.open(method,endpoint,true);
 
 xhttp.setRequestHeader("X-CSRFToken", csrfToken);
 xhttp.send( JSON.stringify( request ) );
 
 xhttp.onreadystatechange = function() {
 if(xhttp.status == 200){ 
 var response = xhttp.responseText;
 console.log(response);
 ProccessResponse(response);
 } else {
 console.log(xhttp.status);
 }
 }
}
async function LogLikes(){
 
 if(!logInterval){
 //SendRequest("GET","/action", '{"method" : "loglikes"}');
 logInterval = setInterval(function() { SendRequest("POST","/action", '{"method" : "getlikeslog"}') }, 45000);
 document.getElementById("likes-log").scrollTop = document.getElementById("likes-log").scrollHeight;
 }
}
window.onload = function(){
 var $ = function(id){ return document.getElementById(id); }
 csrfToken = document.getElementsByTagName("meta")[0].content;
 
 DisplayMenu("None"); // Hide everything Initally.
 
 // Auto Liker Button
 $("autolike-btn").addEventListener("click",function(){ 
 DisplayMenu("autolike-menu");
 })
 
 // Auto Liker Tag
 $("like-start").addEventListener("click",function(){
 var tag = $("like-tag").value;
 console.log(tag);
 SendRequest("POST","/action", '{"method" : "autolike" , "tag" : "'+tag+'" }');
 $("likes-log").value = "Starting Like Automation on tag: "+tag+"\n";
 LogLikes();
 })
 
 $("like-stop").addEventListener("click",function(){
 SendRequest("POST","/action", '{"method" : "stoplike"}');
 $("likes-log").value = "Stopping Liking...";
 })
 
 // Auto Liker Button
 $("autocomment-btn").addEventListener("click",function(){ 
 DisplayMenu("autocomment-menu");
 })
 
 
}
```

• 200_success: Do you have a sample record of what actually gets POSTed to /action? Alternatively, could you show us the form that leads to the POST to /action (maybe the dashboard.html template)?
• Drake Rose: I made an edit to this post. You can check out my .js code. The window.onload contains the "like-start" event listener that sends the json string to the server.
• 200_success: The SendRequest() function JSON.stringify()s the request parameter, but in LogLikes(), you're calling SendRequest() with a third argument that is already a string — '{"method" : "getlikeslog"}'. If you don't double-stringify, then you shouldn't have to double-decode.
• Drake Rose: Thanks for pointing that out I'll make that change. I just need someone to review my code in python and javascript and determine what kind of improvements I should make.

Answer 1

I haven't written JavaScript or dealt with Flask for a couple years, but here are some suggestions for the Python code:

1. black can automatically format your code to be more idiomatic.
2. isort can group and sort your imports automatically.

3. flake8 with a strict complexity limit will give you more hints to write idiomatic Python:

   [flake8]
   max-complexity = 4
   ignore = W503,E203
   

   That limit is not absolute by any means, but it's worth thinking hard whether you can keep it low whenever validation fails. For example, I'm working with a team on an application since a year now, and our complexity limit is up to 7 in only one place. In your case process_action is a good place to start pulling things apart.

4. I would then recommend adding type hints everywhere and validating them using a strict mypy configuration:

   [mypy]
   check_untyped_defs = true
   disallow_untyped_defs = true
   ignore_missing_imports = true
   no_implicit_optional = true
   warn_redundant_casts = true
   warn_return_any = true
   warn_unused_ignores = true
   

Assuming the above have been taken care of:

1. Variable naming is incredibly important for maintainability. Unfortunately vacuous names like data are common, but I would recommend thinking about them and renaming them to something which is readable. In this case something like request_parameters might be appropriate.
2. Having two routes to the main page is a bit confusing. I would suggest avoiding future headaches by having only a single canonical URL per page.
3. http.HTTPStatus contains values for HTTP status codes, which is more readable than magic values.
4. Initializing a variable with list() rather than [] is unusual. I don't expect it will actually make any practical difference, but it does stand out a bit.
5. Rather than printing I would recommend using logging instead - it's much more flexible for when you want to move to more production-like log handling.
6. json.dumps is preferable to manual string building. It handles things like escaping for you, and also makes it easy for you to deal with Python dicts and lists until the very end.

7. Try to aim for less nesting where possible. Returning early is one way to achieve this. For example,

   if 'user' in session:
    [lots of indented code]
   else:
    return redirect(url_for("index"))
   

   could instead be written as

   if 'user' not in session:
    return redirect(url_for("index"))
   [lots of dedented code]
   

   . The same technique can be easily applied to validate_username.

8. time.sleep is not going to be accepted in production code. Not even for a millisecond. If for whatever reason things don't work when running with the breaks off, debug until it works with the breaks off.
9. You have some magic values like 45 in your code. Since it's not obvious what they mean they should either be changed to refer to some third-party constant or be pulled out to a variable or constant clearly indicating what they are.
10. username[len(username) - 1] can be written username[-1].

11. validate_username is either very buggy or extremely confusing. First, it returns either True, False or None, but a method like this should never rely on type coercion. Second, it returns True if a bunch of things are wrong with the username, False in case of ValueError, and None in every other case. Third, since

    >>> None is False
    False
    

    almost every weird username will be accepted as valid.

In conclusion: like most Python code this could benefit from common language patterns and a thorough set of tests.

• Drake Rose: Wow, this was a very in-depth examination of my code I'm really grateful that you took apart and busted it open. I can see so many things I need to improve now! I'll apply all the fixes you suggested. :)