vassili/seitime-frappe
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 4

Merge branch 'develop' into fix-grid-validation

Browse source
This commit is contained in:
Prssanna Desai 2021-03-31 22:38:19 +05:30 • committed by GitHub
commit 0ac2a1f772
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
35 changed files with 295 additions and 189 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

25
frappe/__init__.py
View file

@ -555,8 +555,15 @@ def whitelist(allow_guest=False, xss_safe=False, methods=None):
def innerfn(fn):
global whitelisted, guest_methods, xss_safe_methods, allowed_http_methods_for_whitelisted_func
whitelisted.append(fn)
# get function from the unbound / bound method
# this is needed because functions can be compared, but not methods
method = None
if hasattr(fn, '__func__'):
method = fn
fn = method.__func__
whitelisted.append(fn)
allowed_http_methods_for_whitelisted_func[fn] = methods
if allow_guest:
@ -565,10 +572,24 @@ def whitelist(allow_guest=False, xss_safe=False, methods=None):
if xss_safe:
xss_safe_methods.append(fn)
return fn
return method or fn
return innerfn
def is_whitelisted(method):
from frappe.utils import sanitize_html
is_guest = session['user'] == 'Guest'
if method not in whitelisted or is_guest and method not in guest_methods:
throw(_("Not permitted"), PermissionError)
if is_guest and method not in xss_safe_methods:
# strictly sanitize form_dict
# escapes html characters like <> except for predefined tags like a, b, ul etc.
for key, value in form_dict.items():
if isinstance(value, string_types):
form_dict[key] = sanitize_html(value)
def read_only():
def innfn(fn):
def wrapper_fn(*args, **kwargs):

49
frappe/auth.py
View file

@ -215,35 +215,25 @@ class LoginManager:
if not (user and pwd):
self.fail(_('Incomplete login details'), user=user)
# Ignore password check if tmp_id is set, 2FA takes care of authentication.
validate_password = not bool(frappe.form_dict.get('tmp_id'))
user = User.find_by_credentials(user, pwd, validate_password=validate_password)
user = User.find_by_credentials(user, pwd)
if not user:
self.fail('Invalid login credentials')
sys_settings = frappe.get_doc("System Settings")
track_login_attempts = (sys_settings.allow_consecutive_login_attempts >0)
tracker_kwargs = {}
if track_login_attempts:
tracker_kwargs['lock_interval'] = sys_settings.allow_login_after_fail
tracker_kwargs['max_consecutive_login_attempts'] = sys_settings.allow_consecutive_login_attempts
tracker = LoginAttemptTracker(user.name, **tracker_kwargs)
if track_login_attempts and not tracker.is_user_allowed():
frappe.throw(_("Your account has been locked and will resume after {0} seconds")
.format(sys_settings.allow_login_after_fail), frappe.SecurityException)
# Current login flow uses cached credentials for authentication while checking OTP.
# Incase of OTP check, tracker for auth needs to be disabled(If not, it can remove tracker history as it is going to succeed anyway)
# Tracker is activated for 2FA incase of OTP.
ignore_tracker = should_run_2fa(user.name) and ('otp' in frappe.form_dict)
tracker = None if ignore_tracker else get_login_attempt_tracker(user.name)
if not user.is_authenticated:
tracker.add_failure_attempt()
tracker and tracker.add_failure_attempt()
self.fail('Invalid login credentials', user=user.name)
elif not (user.name == 'Administrator' or user.enabled):
tracker.add_failure_attempt()
tracker and tracker.add_failure_attempt()
self.fail('User disabled or missing', user=user.name)
else:
tracker.add_success_attempt()
tracker and tracker.add_success_attempt()
self.user = user.name
def force_user_to_reset_password(self):
@ -406,6 +396,27 @@ def validate_ip_address(user):
frappe.throw(_("Access not allowed from this IP Address"), frappe.AuthenticationError)
def get_login_attempt_tracker(user_name: str, raise_locked_exception: bool = True):
"""Get login attempt tracker instance.
:param user_name: Name of the loggedin user
:param raise_locked_exception: If set, raises an exception incase of user not allowed to login
"""
sys_settings = frappe.get_doc("System Settings")
track_login_attempts = (sys_settings.allow_consecutive_login_attempts >0)
tracker_kwargs = {}
if track_login_attempts:
tracker_kwargs['lock_interval'] = sys_settings.allow_login_after_fail
tracker_kwargs['max_consecutive_login_attempts'] = sys_settings.allow_consecutive_login_attempts
tracker = LoginAttemptTracker(user_name, **tracker_kwargs)
if raise_locked_exception and track_login_attempts and not tracker.is_user_allowed():
frappe.throw(_("Your account has been locked and will resume after {0} seconds")
.format(sys_settings.allow_login_after_fail), frappe.SecurityException)
return tracker
class LoginAttemptTracker(object):
"""Track login attemts of a user.

2
frappe/automation/doctype/auto_repeat/auto_repeat.py
View file

@ -118,6 +118,7 @@ class AutoRepeat(Document):
def is_completed(self):
return self.end_date and getdate(self.end_date) < getdate(today())
@frappe.whitelist()
def get_auto_repeat_schedule(self):
schedule_details = []
start_date = getdate(self.start_date)
@ -328,6 +329,7 @@ class AutoRepeat(Document):
make(doctype=new_doc.doctype, name=new_doc.name, recipients=recipients,
subject=subject, content=message, attachments=attachments, send_email=1)
@frappe.whitelist()
def fetch_linked_contacts(self):
if self.reference_doctype and self.reference_document:
res = get_contacts_linking_to(self.reference_doctype, self.reference_document, fields=['email_id'])

16
frappe/client.py
View file

@ -21,7 +21,7 @@ Requests via FrappeClient are also handled here.
@frappe.whitelist()
def get_list(doctype, fields=None, filters=None, order_by=None,
limit_start=None, limit_page_length=20, parent=None):
limit_start=None, limit_page_length=20, parent=None, debug=False, as_dict=True):
'''Returns a list of records by filters, fields, ordering and limit
:param doctype: DocType of the data to be queried
@ -40,10 +40,11 @@ def get_list(doctype, fields=None, filters=None, order_by=None,
order_by=order_by,
limit_start=limit_start,
limit_page_length=limit_page_length,
debug=debug,
as_list=not as_dict
)
validate_args(args)
return frappe.get_list(**args)
@frappe.whitelist()
@ -103,14 +104,15 @@ def get_value(doctype, fieldname, filters=None, as_dict=True, debug=False, paren
if frappe.get_meta(doctype).issingle:
value = frappe.db.get_values_from_single(fields, filters, doctype, as_dict=as_dict, debug=debug)
else:
value = get_list(doctype, filters=filters, fields=fields, limit_page_length=1)
value = get_list(doctype, filters=filters, fields=fields, debug=debug, limit_page_length=1, as_dict=as_dict)
if as_dict:
value = value[0] if value else {}
else:
value = value[0][fieldname]
return value[0] if value else {}
return value
if not value:
return
return value[0] if len(fields) > 1 else value[0][0]
@frappe.whitelist()
def get_single_value(doctype, field):

1
frappe/core/doctype/data_import/data_import.py
View file

@ -38,6 +38,7 @@ class DataImport(Document):
return
validate_google_sheets_url(self.google_sheets_url)
@frappe.whitelist()
def get_preview_from_template(self, import_file=None, google_sheets_url=None):
if import_file:
self.import_file = import_file

3
frappe/core/doctype/report/report.py
View file

@ -58,6 +58,7 @@ class Report(Document):
def get_columns(self):
return [d.as_dict(no_default_fields = True) for d in self.columns]
@frappe.whitelist()
def set_doctype_roles(self):
if not self.get('roles') and self.is_standard == 'No':
meta = frappe.get_meta(self.ref_doctype)
@ -304,7 +305,7 @@ class Report(Document):
return data
@Document.whitelist
@frappe.whitelist()
def toggle_disable(self, disable):
self.db_set("disabled", cint(disable))

3
frappe/core/doctype/role_permission_for_page_and_report/role_permission_for_page_and_report.py
View file

@ -8,6 +8,7 @@ from frappe.core.doctype.report.report import is_prepared_report_disabled
from frappe.model.document import Document
class RolePermissionforPageandReport(Document):
@frappe.whitelist()
def set_report_page_data(self):
self.set_custom_roles()
self.check_prepared_report_disabled()
@ -35,12 +36,14 @@ class RolePermissionforPageandReport(Document):
doc = frappe.get_doc(doctype, docname)
return doc.roles
@frappe.whitelist()
def reset_roles(self):
roles = self.get_standard_roles()
self.set('roles', roles)
self.update_custom_roles()
self.update_disable_prepared_report()
@frappe.whitelist()
def update_report_page_data(self):
self.update_custom_roles()
self.update_disable_prepared_report()

2
frappe/core/doctype/user/user.py
View file

@ -558,7 +558,7 @@ class User(Document):
user['is_authenticated'] = True
if validate_password:
try:
check_password(user['name'], password)
check_password(user['name'], password, delete_tracker_cache=False)
except frappe.AuthenticationError:
user['is_authenticated'] = False

3
frappe/custom/doctype/customize_form/customize_form.py
View file

@ -24,6 +24,7 @@ class CustomizeForm(Document):
frappe.db.sql("delete from tabSingles where doctype='Customize Form'")
frappe.db.sql("delete from `tabCustomize Form Field`")
@frappe.whitelist()
def fetch_to_customize(self):
self.clear_existing_doc()
if not self.doc_type:
@ -133,6 +134,7 @@ class CustomizeForm(Document):
self.doc_type = doc_type
self.name = "Customize Form"
@frappe.whitelist()
def save_customization(self):
if not self.doc_type:
return
@ -448,6 +450,7 @@ class CustomizeForm(Document):
self.flags.update_db = True
@frappe.whitelist()
def reset_to_defaults(self):
if not self.doc_type:
return

1
frappe/data_migration/doctype/data_migration_run/data_migration_run.py
View file

@ -10,6 +10,7 @@ from frappe.utils import cstr
from frappe.data_migration.doctype.data_migration_mapping.data_migration_mapping import get_source_value
class DataMigrationRun(Document):
@frappe.whitelist()
def run(self):
self.begin()
if self.total_pages > 0:

81
frappe/desk/form/run_method.py
View file

@ -1,81 +0,0 @@
# Copyright (c) 2015, Frappe Technologies Pvt. Ltd. and Contributors
# MIT License. See license.txt
from __future__ import unicode_literals
import json, inspect
import frappe
from frappe import _
from frappe.utils import cint
from six import text_type, string_types
@frappe.whitelist()
def runserverobj(method, docs=None, dt=None, dn=None, arg=None, args=None):
"""run controller method - old style"""
if not args: args = arg or ""
if dt: # not called from a doctype (from a page)
if not dn: dn = dt # single
doc = frappe.get_doc(dt, dn)
else:
doc = frappe.get_doc(json.loads(docs))
doc._original_modified = doc.modified
doc.check_if_latest()
if not doc.has_permission("read"):
frappe.msgprint(_("Not permitted"), raise_exception = True)
if doc:
try:
args = json.loads(args)
except ValueError:
args = args
try:
fnargs, varargs, varkw, defaults = inspect.getargspec(getattr(doc, method))
except ValueError:
fnargs = inspect.getfullargspec(getattr(doc, method)).args
varargs = inspect.getfullargspec(getattr(doc, method)).varargs
varkw = inspect.getfullargspec(getattr(doc, method)).varkw
defaults = inspect.getfullargspec(getattr(doc, method)).defaults
if not fnargs or (len(fnargs)==1 and fnargs[0]=="self"):
r = doc.run_method(method)
elif "args" in fnargs or not isinstance(args, dict):
r = doc.run_method(method, args)
else:
r = doc.run_method(method, **args)
if r:
#build output as csv
if cint(frappe.form_dict.get('as_csv')):
make_csv_output(r, doc.doctype)
else:
frappe.response['message'] = r
frappe.response.docs.append(doc)
def make_csv_output(res, dt):
"""send method response as downloadable CSV file"""
import frappe
from six import StringIO
import csv
f = StringIO()
writer = csv.writer(f)
for r in res:
row = []
for v in r:
if isinstance(v, string_types):
v = v.encode("utf-8")
row.append(v)
writer.writerow(row)
f.seek(0)
frappe.response['result'] = text_type(f.read(), 'utf-8')
frappe.response['type'] = 'csv'
frappe.response['doctype'] = dt.replace(' ','')

5
frappe/desk/search.py
View file

@ -6,8 +6,7 @@ from __future__ import unicode_literals
import frappe, json
from frappe.utils import cstr, unique, cint
from frappe.permissions import has_permission
from frappe.handler import is_whitelisted
from frappe import _
from frappe import _, is_whitelisted
from six import string_types
import re
import wrapt
@ -221,4 +220,4 @@ def validate_and_sanitize_search_inputs(fn, instance, args, kwargs):
if kwargs['doctype'] and not frappe.db.exists('DocType', kwargs['doctype']):
return []
return fn(**kwargs)
return fn(**kwargs)

1
frappe/email/doctype/newsletter/newsletter.py
View file

@ -29,6 +29,7 @@ class Newsletter(WebsiteGenerator):
self.queue_all(test_email=True)
frappe.msgprint(_("Test email sent to {0}").format(self.test_email_id))
@frappe.whitelist()
def send_emails(self):
"""send emails to leads and customers"""
if self.email_sent:

114
frappe/handler.py
View file

@ -2,17 +2,19 @@
# MIT License. See license.txt
from __future__ import unicode_literals
from werkzeug.wrappers import Response
import frappe
from frappe import _
import frappe.utils
import frappe.sessions
import frappe.desk.form.run_method
from frappe.utils.response import build_response
from frappe.api import validate_auth
from frappe.utils import cint
from frappe.api import validate_auth
from frappe import _, is_whitelisted
from frappe.utils.response import build_response
from frappe.utils.csvutils import build_csv_response
from frappe.core.doctype.server_script.server_script_utils import run_server_script_api
from werkzeug.wrappers import Response
from six import string_types
ALLOWED_MIMETYPES = ('image/png', 'image/jpeg', 'application/pdf', 'application/msword',
'application/vnd.openxmlformats-officedocument.wordprocessingml.document',
@ -64,8 +66,9 @@ def execute_cmd(cmd, from_async=False):
if from_async:
method = method.queue
is_whitelisted(method)
is_valid_http_method(method)
if method != run_doc_method:
is_whitelisted(method)
is_valid_http_method(method)
return frappe.call(method, **frappe.form_dict)
@ -73,33 +76,15 @@ def is_valid_http_method(method):
http_method = frappe.local.request.method
if http_method not in frappe.allowed_http_methods_for_whitelisted_func[method]:
frappe.throw(_("Not permitted"), frappe.PermissionError)
throw_permission_error()
def is_whitelisted(method):
# check if whitelisted
if frappe.session['user'] == 'Guest':
if (method not in frappe.guest_methods):
frappe.throw(_("Not permitted"), frappe.PermissionError)
if method not in frappe.xss_safe_methods:
# strictly sanitize form_dict
# escapes html characters like <> except for predefined tags like a, b, ul etc.
for key, value in frappe.form_dict.items():
if isinstance(value, string_types):
frappe.form_dict[key] = frappe.utils.sanitize_html(value)
else:
if not method in frappe.whitelisted:
frappe.throw(_("Not permitted"), frappe.PermissionError)
def throw_permission_error():
frappe.throw(_("Not permitted"), frappe.PermissionError)
@frappe.whitelist(allow_guest=True)
def version():
return frappe.__version__
@frappe.whitelist()
def runserverobj(method, docs=None, dt=None, dn=None, arg=None, args=None):
frappe.desk.form.run_method.runserverobj(method, docs=docs, dt=dt, dn=dn, arg=arg, args=args)
@frappe.whitelist(allow_guest=True)
def logout():
frappe.local.login_manager.logout()
@ -112,15 +97,6 @@ def web_logout():
frappe.respond_as_web_page(_("Logged Out"), _("You have been successfully logged out"),
indicator_color='green')
@frappe.whitelist(allow_guest=True)
def run_custom_method(doctype, name, custom_method):
"""cmd=run_custom_method&doctype={doctype}&name={name}&custom_method={custom_method}"""
doc = frappe.get_doc(doctype, name)
if getattr(doc, custom_method, frappe._dict()).is_whitelisted:
frappe.call(getattr(doc, custom_method), **frappe.local.form_dict)
else:
frappe.throw(_("Not permitted"), frappe.PermissionError)
@frappe.whitelist()
def uploadfile():
ret = None
@ -222,6 +198,66 @@ def get_attr(cmd):
frappe.log("method:" + cmd)
return method
@frappe.whitelist(allow_guest = True)
@frappe.whitelist(allow_guest=True)
def ping():
return "pong"
def run_doc_method(method, docs=None, dt=None, dn=None, arg=None, args=None):
"""run a whitelisted controller method"""
import json
import inspect
if not args:
args = arg or ""
if dt: # not called from a doctype (from a page)
if not dn:
dn = dt # single
doc = frappe.get_doc(dt, dn)
else:
doc = frappe.get_doc(json.loads(docs))
doc._original_modified = doc.modified
doc.check_if_latest()
if not doc or not doc.has_permission("read"):
throw_permission_error()
try:
args = json.loads(args)
except ValueError:
args = args
method_obj = getattr(doc, method)
fn = getattr(method_obj, '__func__', method_obj)
is_whitelisted(fn)
is_valid_http_method(fn)
try:
fnargs = inspect.getargspec(method_obj)[0]
except ValueError:
fnargs = inspect.getfullargspec(method_obj).args
if not fnargs or (len(fnargs)==1 and fnargs[0]=="self"):
response = doc.run_method(method)
elif "args" in fnargs or not isinstance(args, dict):
response = doc.run_method(method, args)
else:
response = doc.run_method(method, **args)
frappe.response.docs.append(doc)
if not response:
return
# build output as csv
if cint(frappe.form_dict.get('as_csv')):
build_csv_response(response, doc.doctype.replace(' ', ''))
return
frappe.response['message'] = response
# for backwards compatibility
runserverobj = run_doc_method

1
frappe/integrations/doctype/connected_app/connected_app.py
View file

@ -44,6 +44,7 @@ class ConnectedApp(Document):
scope=self.get_scopes()
)
@frappe.whitelist()
def initiate_web_application_flow(self, user=None, success_uri=None):
"""Return an authorization URL for the user. Save state in Token Cache."""
user = user or frappe.session.user

1
frappe/integrations/doctype/social_login_key/social_login_key.py
View file

@ -49,6 +49,7 @@ class SocialLoginKey(Document):
icon_file = icon_map[self.provider_name]
self.icon = '/assets/frappe/icons/social/{0}'.format(icon_file)
@frappe.whitelist()
def get_social_login_provider(self, provider, initialize=False):
providers = {}

20
frappe/model/document.py
View file

@ -4,7 +4,7 @@
from __future__ import unicode_literals, print_function
import frappe
import time
from frappe import _, msgprint
from frappe import _, msgprint, is_whitelisted
from frappe.utils import flt, cstr, now, get_datetime_str, file_lock, date_diff
from frappe.model.base_document import BaseDocument, get_controller
from frappe.model.naming import set_new_name
@ -126,10 +126,10 @@ class Document(BaseDocument):
raise ValueError('Illegal arguments')
@staticmethod
def whitelist(f):
def whitelist(fn):
"""Decorator: Whitelist method to be called remotely via REST API."""
f.whitelisted = True
return f
frappe.whitelist()(fn)
return fn
def reload(self):
"""Reload document from database"""
@ -1148,12 +1148,12 @@ class Document(BaseDocument):
return composer
def is_whitelisted(self, method):
fn = getattr(self, method, None)
if not fn:
raise NotFound("Method {0} not found".format(method))
elif not getattr(fn, "whitelisted", False):
raise Forbidden("Method {0} not whitelisted".format(method))
def is_whitelisted(self, method_name):
method = getattr(self, method_name, None)
if not method:
raise NotFound("Method {0} not found".format(method_name))
is_whitelisted(getattr(method, '__func__', method))
def validate_value(self, fieldname, condition, val2, doc=None, raise_exception=None):
"""Check that value of fieldname should be 'condition' val2

16
frappe/public/js/frappe/db.js
View file

@ -100,17 +100,11 @@ frappe.db = {
const fields = [];
return frappe.call({
type: 'GET',
method: 'frappe.desk.reportview.get_count',
args: {
doctype,
filters,
fields,
distinct,
}
}).then(r => {
return r.message.values;
return frappe.xcall('frappe.desk.reportview.get_count', {
doctype,
filters,
fields,
distinct,
});
},
get_link_options(doctype, txt = '', filters={}) {

2
frappe/public/js/frappe/form/controls/button.js
View file

@ -34,7 +34,7 @@ frappe.ui.form.ControlButton = frappe.ui.form.ControlData.extend({
var me = this;
if(this.frm && this.frm.docname) {
frappe.call({
method: "runserverobj",
method: "run_doc_method",
args: {'docs': this.frm.doc, 'method': this.df.options },
btn: this.$input,
callback: function(r) {

2
frappe/public/js/frappe/form/multi_select_dialog.js
View file

@ -251,7 +251,6 @@ frappe.ui.form.MultiSelectDialog = class MultiSelectDialog {
head ? $row.addClass('list-item--head')
: $row = $(`<div class="list-item-container" data-item-name="${result.name}"></div>`).append($row);
$(".modal-dialog .list-item--head").css("z-index", 0);
return $row;
}
@ -264,6 +263,7 @@ frappe.ui.form.MultiSelectDialog = class MultiSelectDialog {
this.empty_list();
}
more_btn.hide();
$(".modal-dialog .list-item--head").css("z-index", 0);
if (results.length === 0) return;
if (more) more_btn.show();

2
frappe/public/js/frappe/request.js
View file

@ -55,7 +55,7 @@ frappe.call = function(opts) {
args.cmd = opts.module+'.page.'+opts.page+'.'+opts.page+'.'+opts.method;
} else if(opts.doc) {
$.extend(args, {
cmd: "runserverobj",
cmd: "run_doc_method",
docs: frappe.get_doc(opts.doc.doctype, opts.doc.name),
method: opts.method,
args: opts.args,

2
frappe/public/js/frappe/router.js
View file

@ -364,7 +364,7 @@ frappe.router = {
// return clean sub_path from hash or url
// supports both v1 and v2 routing
if (!route) {
route = window.location.hash || window.location.pathname;
route = window.location.hash || (window.location.pathname + window.location.search);
}
return this.strip_prefix(route);

1
frappe/public/scss/desk/list.scss
View file

@ -261,7 +261,6 @@ input.list-check-all, input.list-row-checkbox {
input[type=checkbox] {
margin: 0;
margin-right: 5px;
flex: 0 0 12px;
}
.liked-by, .liked-by-filter-button {

7
frappe/sessions.py
View file

@ -296,8 +296,7 @@ class Session:
expiry = get_expiry_in_seconds(session_data.get("session_expiry"))
if self.time_diff > expiry:
print('deleting...')
self.delete_session()
self._delete_session()
data = None
return data and data.data
@ -316,12 +315,12 @@ class Session:
data = frappe._dict(eval(rec and rec[0][1] or '{}'))
data.user = rec[0][0]
else:
self.delete_session()
self._delete_session()
data = None
return data
def delete_session(self):
def _delete_session(self):
delete_session(self.sid, reason="Session Expired")
def start_as_guest(self):

1
frappe/social/doctype/energy_point_log/energy_point_log.py
View file

@ -52,6 +52,7 @@ class EnergyPointLog(Document):
reference_log.reverted = 0
reference_log.save()
@frappe.whitelist()
def revert(self, reason, ignore_permissions=False):
if not ignore_permissions:
frappe.only_for('System Manager')

9
frappe/tests/__init__.py
View file

@ -0,0 +1,9 @@
import frappe
def update_system_settings(args):
doc = frappe.get_doc('System Settings')
doc.update(args)
doc.save()
def get_system_setting(key):
return frappe.db.get_single_value("System Settings", key)

3
frappe/tests/test_api.py
View file

@ -143,8 +143,7 @@ class TestAPI(unittest.TestCase):
self.assertFalse(frappe.db.get_value('Note', {'title': 'delete'}))
def test_auth_via_api_key_secret(self):
# generate api ke and api secret for administrator
# generate API key and API secret for administrator
keys = generate_keys("Administrator")
frappe.db.commit()
generated_secret = frappe.utils.password.get_decrypted_password(

50
frappe/tests/test_client.py
View file

@ -2,7 +2,9 @@
from __future__ import unicode_literals
import unittest, frappe
import unittest
import frappe
class TestClient(unittest.TestCase):
def test_set_value(self):
@ -55,3 +57,49 @@ class TestClient(unittest.TestCase):
})
self.assertRaises(frappe.PermissionError, execute_cmd, 'frappe.client.save')
def test_run_doc_method(self):
from frappe.handler import execute_cmd
if not frappe.db.exists('Report', 'Test Run Doc Method'):
report = frappe.get_doc({
'doctype': 'Report',
'ref_doctype': 'User',
'report_name': 'Test Run Doc Method',
'report_type': 'Query Report',
'is_standard': 'No',
'roles': [
{'role': 'System Manager'}
]
}).insert()
else:
report = frappe.get_doc('Report', 'Test Run Doc Method')
frappe.local.request = frappe._dict()
frappe.local.request.method = 'GET'
# Whitelisted, works as expected
frappe.local.form_dict = frappe._dict({
'dt': report.doctype,
'dn': report.name,
'method': 'toggle_disable',
'cmd': 'run_doc_method',
'args': 0
})
execute_cmd(frappe.local.form_dict.cmd)
# Not whitelisted, throws permission error
frappe.local.form_dict = frappe._dict({
'dt': report.doctype,
'dn': report.name,
'method': 'create_report_py',
'cmd': 'run_doc_method',
'args': 0
})
self.assertRaises(
frappe.PermissionError,
execute_cmd,
frappe.local.form_dict.cmd
)

40
frappe/tests/test_twofactor.py
View file

@ -6,23 +6,34 @@ import unittest, frappe, pyotp
from frappe.auth import HTTPRequest
from frappe.utils import cint
from frappe.utils import set_request
from frappe.auth import validate_ip_address
from frappe.auth import validate_ip_address, get_login_attempt_tracker
from frappe.twofactor import (should_run_2fa, authenticate_for_2factor, get_cached_user_pass,
two_factor_is_enabled_for_, confirm_otp_token, get_otpsecret_for_, get_verification_obj)
from . import update_system_settings, get_system_setting
import time
class TestTwoFactor(unittest.TestCase):
def __init__(self, *args, **kwargs):
super(TestTwoFactor, self).__init__(*args, **kwargs)
self.default_allowed_login_attempts = get_system_setting('allow_consecutive_login_attempts')
def setUp(self):
self.http_requests = create_http_request()
self.login_manager = frappe.local.login_manager
self.user = self.login_manager.user
update_system_settings({
'allow_consecutive_login_attempts': 2
})
def tearDown(self):
frappe.local.response['verification'] = None
frappe.local.response['tmp_id'] = None
disable_2fa()
frappe.clear_cache(user=self.user)
update_system_settings({
'allow_consecutive_login_attempts': self.default_allowed_login_attempts
})
def test_should_run_2fa(self):
'''Should return true if enabled.'''
@ -153,6 +164,33 @@ class TestTwoFactor(unittest.TestCase):
enable_2fa()
self.assertIsNone(validate_ip_address(self.user))
def test_otp_attempt_tracker(self):
"""Check that OTP login attempts are tracked.
"""
authenticate_for_2factor(self.user)
tmp_id = frappe.local.response['tmp_id']
otp = 'wrongotp'
with self.assertRaises(frappe.AuthenticationError):
confirm_otp_token(self.login_manager,otp=otp,tmp_id=tmp_id)
with self.assertRaises(frappe.AuthenticationError):
confirm_otp_token(self.login_manager,otp=otp,tmp_id=tmp_id)
# REMOVE ME: current logic allows allow_consecutive_login_attempts+1 attempts
# before raising security exception, remove below line when that is fixed.
with self.assertRaises(frappe.AuthenticationError):
confirm_otp_token(self.login_manager,otp=otp,tmp_id=tmp_id)
with self.assertRaises(frappe.SecurityException):
confirm_otp_token(self.login_manager,otp=otp,tmp_id=tmp_id)
# Remove tracking cache so that user can try loging in again
tracker = get_login_attempt_tracker(self.user, raise_locked_exception=False)
tracker.add_success_attempt()
otp = get_otp(self.user)
self.assertTrue(confirm_otp_token(self.login_manager,otp=otp,tmp_id=tmp_id))
def create_http_request():
'''Get http request object.'''
set_request(method='POST', path='login')

8
frappe/twofactor.py
View file

@ -118,6 +118,7 @@ def get_verification_method():
def confirm_otp_token(login_manager, otp=None, tmp_id=None):
'''Confirm otp matches.'''
from frappe.auth import get_login_attempt_tracker
if not otp:
otp = frappe.form_dict.get('otp')
if not otp:
@ -130,12 +131,17 @@ def confirm_otp_token(login_manager, otp=None, tmp_id=None):
otp_secret = frappe.cache().get(tmp_id + '_otp_secret')
if not otp_secret:
raise ExpiredLoginException(_('Login session expired, refresh page to retry'))
tracker = get_login_attempt_tracker(login_manager.user)
hotp = pyotp.HOTP(otp_secret)
if hotp_token:
if hotp.verify(otp, int(hotp_token)):
frappe.cache().delete(tmp_id + '_token')
tracker.add_success_attempt()
return True
else:
tracker.add_failure_attempt()
login_manager.fail(_('Incorrect Verification code'), login_manager.user)
totp = pyotp.TOTP(otp_secret)
@ -144,8 +150,10 @@ def confirm_otp_token(login_manager, otp=None, tmp_id=None):
if not frappe.db.get_default(login_manager.user + '_otplogin'):
frappe.db.set_default(login_manager.user + '_otplogin', 1)
delete_qrimage(login_manager.user)
tracker.add_success_attempt()
return True
else:
tracker.add_failure_attempt()
login_manager.fail(_('Incorrect Verification code'), login_manager.user)

8
frappe/utils/password.py
View file

@ -65,7 +65,7 @@ def set_encrypted_password(doctype, name, pwd, fieldname='password'):
raise e
def check_password(user, pwd, doctype='User', fieldname='password'):
def check_password(user, pwd, doctype='User', fieldname='password', delete_tracker_cache=True):
'''Checks if user and password are correct, else raises frappe.AuthenticationError'''
auth = frappe.db.sql("""select `name`, `password` from `__Auth`
@ -77,7 +77,11 @@ def check_password(user, pwd, doctype='User', fieldname='password'):
# lettercase agnostic
user = auth[0].name
delete_login_failed_cache(user)
# TODO: This need to be deleted after checking side effects of it.
# We have a `LoginAttemptTracker` that can take care of tracking related cache.
if delete_tracker_cache:
delete_login_failed_cache(user)
if not passlibctx.needs_update(auth[0].password):
update_password(user, pwd, doctype, fieldname)

1
frappe/website/doctype/blog_post/blog_post.py
View file

@ -18,6 +18,7 @@ class BlogPost(WebsiteGenerator):
order_by = "published_on desc"
)
@frappe.whitelist()
def make_route(self):
if not self.route:
return frappe.db.get_value('Blog Category', self.blog_category,

1
frappe/website/doctype/personal_data_deletion_request/personal_data_deletion_request.py
View file

@ -101,6 +101,7 @@ class PersonalDataDeletionRequest(Document):
if self.status != "Pending Approval":
frappe.throw(_("This request has not yet been approved by the user."))
@frappe.whitelist()
def trigger_data_deletion(self):
"""Redact user data defined in current site's hooks under `user_data_fields`"""
self.validate_data_anonymization()

1
frappe/website/doctype/portal_settings/portal_settings.py
View file

@ -19,6 +19,7 @@ class PortalSettings(Document):
self.append('menu', item)
return True
@frappe.whitelist()
def reset(self):
'''Restore defaults'''
self.menu = []

2
frappe/website/doctype/website_theme/website_theme.py
View file

@ -98,6 +98,7 @@ class WebsiteTheme(Document):
else:
self.generate_bootstrap_theme()
@frappe.whitelist()
def set_as_default(self):
self.generate_bootstrap_theme()
self.save()
@ -106,6 +107,7 @@ class WebsiteTheme(Document):
website_settings.ignore_validate = True
website_settings.save()
@frappe.whitelist()
def get_apps(self):
from frappe.utils.change_log import get_versions
apps = get_versions()